{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "本章从零开始逐步实现一个类似muduo的基于Reactor模式的C++网络库，大体反映了muduo网络相关部分的开发过程。本章大致分为三段，为了与代码匹配，本章的小节从0开始编号。注意本章呈现的代码与现在muduo的代码略有出入。\n",
    "\n",
    "1. § 8.0至 § 8.3介绍Reactor模式的现代C++实现，包括`EventLoop`、`Poller`、`Channel`、`TimerQueue`、`EventLoopThread`等class；\n",
    "2. § 8.4至 § 8.9介绍基于Reactor的单线程、非阻塞、并发TCP server网络编程，主要介绍`Acceptor`、`Socket`、`TcpServer`、`TcpConnection`、`Buffer`等class；\n",
    "3. § 8.10至 § 8.13是提高篇，介绍`one loop per thread`的实现（用`EventLoopThreadPool`实现多线程`TcpServer`），`Connector`和`TcpClient` class，还有用`epoll(4)`替换`poll(2)`作为`Poller`的IO multiplexing机制等。\n",
    "\n",
    "本章的代码位于[`recipes/reactor/`](../recipes/reactor/)，会直接使用[`muduo/base`](../muduo/base/)中的日志、线程等基础库。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 8.0 什么都不做的EventLoop\n",
    "\n",
    "首先定义`EventLoop class`的基本接口：构造函数、析构函数、`loop()`成员函数。注意`EventLoo`p是不可拷贝的，因此它继承了[`boost::noncopyable`](../muduo/base/noncopyable.h)。muduo中的大多数class都是不可拷贝的，因此以后只会强调某个class是可拷贝的。\n",
    "> 因为基类把拷贝构造和拷贝赋值运算符声明为 `deleted`，派生类复制时需要先复制基类子对象而被编译器禁止，因此通过继承即可禁止复制。\n",
    "\n",
    "- [EventLoop.h](./00_eventloop/EventLoop.h)\n",
    "\n",
    "one loop per thread顾名思义每个线程只能有一个 `EventLoop` 对象，因此 Event-Loop 的构造函数会检查当前线程是否已经创建了其他 `EventLoop` 对象，遇到错误就终止程序（`LOG_FATAL`）。`EventLoop` 的构造函数会记住本对象所属的线程（`threadId_`）。创建了 `EventLoop` 对象的线程是 IO 线程，其主要功能是运行事件循环 `EventLoop::loop()`。`EventLoop` 对象的生命周期通常和其所属的线程一样长，它不必是 heap 对象。\n",
    "\n",
    "- [EventLoop.cc](./00_eventloop/EventLoop.cc)\n",
    "\n",
    "既然每个线程至多有一个 `EventLoop` 对象，那么我们让 `EventLoop` 的 `static` 成员函数 `getEventLoopOfCurrentThread()` 返回这个对象。返回值可能为 `NULL`，如果当前线程不是 IO 线程的话。（这个函数是 muduo 后来新加的，因此前面头文件中没有它的原型。）\n",
    "\n",
    "- [getEventLoopOfCurrentThread()](./00_eventloop/EventLoop.cc#62)\n",
    "\n",
    "muduo的接口设计会明确哪些成员函数是线程安全的，可以跨线程调用；哪些成员函数只能在某个特定线程调用（主要是IO线程）。为了能在运行时检查这些`pre-condition`，`EventLoop`提供了`isInLoopThread()`和`assertInLoopThread()`等函数（[`EventLoop.h`](./00_eventloop/EventLoop.h#L22)），其中用到的`EventLoop::abortNotInLoopThread()`函数的定义从略。\n",
    "\n",
    "事件循环必须在IO线程执行，因此`EventLoop::loop()`会检查这一[`pre-condition`](./00_eventloop/EventLoop.cc#44)。本节的`loop()`什么事都不做，等5秒就退出。\n",
    "\n",
    "为了验证现有的功能，我编写了[s00/test1.cc](./00_eventloop/test01.cc)和[s00/test2.cc](./00_eventloop/test02.cc)。其中test1.cc会在主线程和子线程分别创建一个EventLoop，程序正常运行退出。\n",
    "\n",
    "`test2.cc`是个负面测试，它在主线程创建了`EventLoop`对象，却[试图](./00_eventloop/test02.cc#L8)在另一个线程调用其`EventLoop::loop()`，程序会因断言失效而异常终"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "main(): pid = 7198, tid = 7198\n",
      "threadFunc(): pid = 7198, tid = 7199\n"
     ]
    }
   ],
   "source": [
    "! cd ../run && ./00_eventloop_test01"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "20251015 06:07:00.629039Z  7224 FATAL EventLoop::abortNotInLoopThread - EventLoop 0x7FFEC289C0D8 was created in threadId_ = 7223, current thread id = 7224 - EventLoop.cpp:56\n",
      "Aborted\n"
     ]
    }
   ],
   "source": [
    "! cd ../run && ./00_eventloop_test02"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 练习：写一个负面测试，在主线程创建两个`EventLoop`对象，验证程序会异常终止。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "20251015 06:07:18.919734Z  7287 FATAL EventLoop::abortNotInLoopThread - EventLoop 0x7FFE16770EE0 was created in threadId_ = 7286, current thread id = 7287 - EventLoop.cpp:56\n",
      "Aborted\n"
     ]
    }
   ],
   "source": [
    "! cd ../run && ./00_eventloop_test03"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 8.1 Reactor的关键结构\n",
    "\n",
    "**TL;DR**\n",
    "\n",
    "- **Channel：** 负责“某个 fd + 感兴趣事件”的打包与回调分发；只属于一个 EventLoop/线程，不拥有 fd，只暴露 `set*Callback()` + `enableReading()` 等接口；`handleEvent()` 根据 `revents_` 调用相应回调；`update()` 交给所属的 EventLoop → Poller。\n",
    "- **Poller：** 封装具体的 I/O 多路复用（此处基于 `poll(2)`）；维护 `pollfds_` 与 `fd → Channel*` 映射；`poll()` 收集活跃事件并填充 `activeChannels`，不直接触发回调；`updateChannel()` 负责新增/更新 Channel 在 `pollfds_` 中的位置。\n",
    "- **EventLoop：** 持有一个 Poller，循环执行 `poll()`、遍历 `activeChannels`、调用 `Channel::handleEvent()`；`quit()` 通过标志安全退出循环。\n",
    "- **调试小结：** Level-trigger（poll）便于用 `strace` 观察重复唤醒和未读完数据的情况；若换成 ET（如 epoll ET），事件只在状态边沿触发，`strace` 仍能看系统调用，但缺少 LT 模式那种“重复唤醒”的直观痕迹，需靠程序自身日志来断言处理是否彻底。\n",
    "\n",
    "\n",
    "\n",
    "本节讲Reactor最核心的事件分发机制，即将IO multiplexing拿到的IO事件分发给各个文件描述符(fd)的事件处理函数。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## `Channel class`\n",
    "\n",
    "`Channel class`的功能有一点类似Java NIO的SelectableChannel和Selection-Key的组合。每个`Channel`对象自始至终只属于一个`EventLoop`，因此每个`Channel`对象都只属于某一个IO线程。每个`Channel`对象自始至终只负责一个文件描述符(fd)的IO事件分发，但它并不拥有这个fd，也不会在析构的时候关闭这个fd。`Channel`会把不同的IO事件分发为不同的回调，例如`ReadCallback`、`WriteCallback`等，而且“回调”用`boost::function`表示，用户无需继承`Channel`，`Channel`不是基类。muduo用户一般不直接使用`Channel`，而会使用更上层的封装，如`TcpConnection`。`Channel`的生命周期由其`owner class`负责管理，它一般是其他class的直接或间接成员。以下是`Channel`的public interface：\n",
    "\n",
    "- [Channel.h](./01_reactor/Channel.h)\n",
    "\n",
    "有些成员函数是内部使用的，用户一般只用`set*Callback()`和`enableReading()`这几个函数。其中有些函数目前还用不到，因此暂时注释起来。`Channel`的成员函数都只能在IO线程调用，因此更新数据成员都不必加锁。\n",
    "\n",
    "以下是`Channel class`的数据成员。其中`events_`是它关心的IO事件，由用户设置；`revents_`是目前活动的事件，由`EventLoop/Poller`设置；这两个字段都是bit pattern，它们的名字来自`poll(2)`的`struct pollfd`。\n",
    "\n",
    "- [Channel.h](./01_reactor/Channel.h#L51)\n",
    "\n",
    "注意到`Channel.h`没有包含任何POSIX头文件，因此`kReadEvent`和`kWriteEvent`等常量的定义要放到[Channel.cc](./01_reactor/Channel.cc)中。\n",
    "\n",
    "[`Channel::update()`](./01_reactor/Channel.cc#L25) 会调用 `EventLoop::updateChannel()`，后者会转而调用 `Poller::updateChannel()`。由于 `Channel.h` 没有包含 `EventLoop.h`，因此 `Channel::update()` 必须定义在 `Channel.cc` 中。\n",
    "\n",
    "[`Channel::handleEvent()`](./01_reactor/Channel.cc#L28) 是 `Channel` 的核心，它由 `EventLoop::loop()` 调用，它的功能是根据 `revents_` 的值分别调用不同的用户回调。这个函数以后还会扩充。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## `Poller class`\n",
    "\n",
    "`Poller class`是IO multiplexing的封装。它现在是个具体类，而在muduo中是个抽象基类，因为muduo同时支持`poll(2)`和`epoll(4)`两种IO multiplexing机制。`Poller`是`EventLoop`的间接成员，只供其owner `EventLoop`在IO线程调用，因此无须加锁。其生命周期与`EventLoop`相等。`Poller`并不拥有`Channel`，`Channel`在析构之前必须自己unregister（`EventLoop::removeChannel()`），避免空悬指针。\n",
    "\n",
    "- [Poller.h](./01_reactor/Poller.h)\n",
    "\n",
    "注意`Poller.h`并没有`include<poll.h>`，而是自己前向声明了`struct pollfd`，这不妨碍我们定义`vector<struct pollfd>`成员。\n",
    "\n",
    "`Poller`供`EventLoop`调用的函数目前有两个，`poll()`和`updateChannel()`，`Poller`暂时没有定义`removeChannel()`成员函数，因为前几节还用不到它。\n",
    "\n",
    "以下是`Poller class`的数据成员。其中`ChannelMap`是从`fd`到`Channel*`的映射。`Poller::poll()`不会在每次调用`poll(2)`之前临时构造`pollfd`数组，而是把它缓存起来（`pollfds_`）。\n",
    "\n",
    "- [Poller.h](./01_reactor/Poller.h#L38)\n",
    "\n",
    "Poller的构造函数和析构函数都很简单，因其成员都是标准库容器。\n",
    "\n",
    "- [Poller.cc](./01_reactor/Poller.cc#L18)\n",
    "\n",
    "`Poller::poll()`是`Poller`的核心功能，它调用`poll(2)`获得当前活动的IO事件，然后填充调用方传入的`activeChannels`，并返回`poll(2)`return的时刻。这里我们直接把`vector<struct pollfd> pollfds_`作为参数传给`poll(2)`，因为C++标准保证`std::vector`的元素排列跟数组一样。[`&*pollfds_.begin()`](./01_reactor/Poller.cc#L23)是获得元素的首地址，这个表达式的类型为`pollfds_*`，符合`poll(2)`的要求。（在`C++11`中可写为`pollfds_.data()`，`g++4.4`的STL也支持这种写法。）\n",
    "\n",
    "- [Poller::poll()](./01_reactor/Poller.cc#L21)\n",
    "\n",
    "`fillActiveChannels()`遍历`pollfds_`，找出有活动事件的`fd`，把它对应的`Channel`填入`activeChannels`。这个函数的复杂度是`O(N)`，其中`N`是`pollfds_`的长度，即文件描述符数目。为了提前结束循环，每找到一个活动`fd`就递减`numEvents`，这样当`numEvents`减为`0`时表示活动`fd`都找完了，不必做无用功。当前活动事件`revents`会保存在`Channel`中，供[`Channel::handleEvent()`](./01_reactor/Channel.cc#L27)使用[（`L47`）](./01_reactor/Poller.cc#L47)。\n",
    "\n",
    "注意这里我们不能一边遍历`pollfds_`，一边调用[`Channel::handleEvent()`](./01_reactor/Channel.cc#L28)，因为后者会添加或删除`Channel`，从而造成`pollfds_`在遍历期间改变大小，这是非常危险的。另一个原因是简化`Poller`的职责，它只负责`IO multiplexing`，不负责事件分发（`dispatching`）。这样将来可以方便地替换为其他更高效的`IO multiplexing`机制，如`epoll(4)`。\n",
    "\n",
    "- [fillActiveChannels()](./01_reactor/Poller.cc#L37)\n",
    "\n",
    "[`Poller::updateChannel()`](./01_reactor/Poller.cc#L55)的主要功能是负责维护和更新`pollfds_`数组。添加新`Channel`的复杂度是`O(log N)`，更新已有的`Channel`的复杂度是`O(1)`，因为`Channel`记住了自己在`pollfds_`数组中的下标，因此可以快速定位。`removeChannel()`的复杂度也将会是`O(log N)`。这里用了大量的`assert`来检查`invariant`。\n",
    "\n",
    "另外，如果某个`Channel`暂时不关心任何事件，就把`pollfd.fd`设为`-1`，[让`poll(2)`忽略此项](./01_reactor/Poller.cc#L83) [注](https://pubs.opengroup.org/onlinepubs/007908799/xsh/poll.html)。这里不能改为把`pollfd.events`设为`0`，这样无法屏蔽`POLLERR`事件。改进的做法（p.312-8.6）是把`pollfd.fd`设为`channel->fd()`的相反数减一，这样可以进一步检查`invariant`。（思考：为什么要减一？）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## EventLoop的改动\n",
    "\n",
    "`EventLoop class`新增了`quit()`成员函数，还加了几个数据成员，并在构造函数里初始化它们。注意`EventLoop`通过`scoped_ptr`来间接持有`Poller`，因此`EventLoop.h`不必包含`Poller.h`，只需前向声明`Poller class`。为此，`EventLoop`的析构函数必须在`EventLoop.cc`中显式定义。\n",
    "\n",
    "- [EventLoop.h](./01_reactor/EventLoop.h#L40)\n",
    "\n",
    "`EventLoop::loop()`有了真正的工作内容，它调用`Poller::poll()`获得当前活动事件的`Channel`列表，然后依次调用每个`Channel`的`handleEvent()`函数。\n",
    "\n",
    "- [EventLoop::loop()](./01_reactor/EventLoop.cc#L49)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "以上几个class尽管简陋，却构成了Reactor模式的核心内容。时序图见图8-1。\n",
    "\n",
    "<img src=\"./images/8.1.png\" alt=\"8.1\" style=\"zoom:40%;\" />"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "我们现在可以终止事件循环，只要将`quit_`设为`true`即可，但是`quit()`不是立刻发生的，它会在`EventLoop::loop()`下一次检查`while(! quit_)`的时候[起效](./01_reactor/EventLoop.cc#L56)。如果在非当前IO线程调用`quit()`，延迟可以长达数秒，将来我们可以唤醒`EventLoop`以缩小延时。但是`quit()`不是中断或`signal`，而是设标志，如果`EventLoop::loop()`正阻塞在某个调用中，[`quit()`](./01_reactor/EventLoop.cc#L71)不会立刻生效。\n",
    "\n",
    "[`EventLoop::updateChannel()`](./01_reactor/EventLoop.cc#L78)在检查断言之后调用`Poller::updateChannel()`，`EventLoop`不关心`Poller`是如何管理`Channel`列表的。\n",
    "\n",
    "有了以上的`EventLoop`、`Poller`、`Channel`，我们写个小程序简单地测试一下功能。[`s01/test3.cc`](./01_reactor/test01.cc)用`timerfd`实现了一个单次触发的定时器，为§ 8.2的内容打下基础。这个程序利用`Channel`将`timerfd`的`readable`事件转发给`timerout()`函数。\n",
    "\n",
    "由于`poll(2)`是level trigger，在`timeout()`中应该`read() timefd`，否则下次会立刻触发。在现阶段采用level trigger的好处之一是可以通过`strace`命令直观地看到每次`poll(2)`的参数列表，容易检查程序的行为。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI:\n",
    "\n",
    "“用 `strace` 观察每次 `poll(2)` 的参数”这个优点，其实是基于 `poll` 天生就是 LT：只要 fd 上的状态一直满足条件（比如 `timerfd` 有未读的超时次数），下一次 `poll` 仍然会立即返回，`strace` 能清楚记录每次系统调用及其参数。\n",
    "\n",
    "如果换成 `epoll` 的 ET 模式情况就不同了：\n",
    "1. **触发条件不同**：ET 只有在状态从“无事件”变为“有事件”的瞬间才触发一次通知（边沿），所以如果你没把缓冲区读空，就不会再次收到通知。LT 则会一直触发，直到状态被你处理完。\n",
    "2. **`strace` 能看到的内容**：`strace` 始终能看到你的系统调用（比如 `epoll_wait`、`read` 等），与 LT/ET 无关。区别在于 ET 模式下，因为事件不会重复触发，你在 `strace` 里看到的 `epoll_wait(..., timeout)` 可能长时间阻塞（甚至返回 0），很难像 LT 那样从连续的 `poll` 返回值中看出“我还有没处理的数据”。换句话说，`strace` 仍然可以用，只是缺少了 LT 模式里“重复唤醒→再次触发→立刻返回”的直观痕迹。\n",
    "3. **调试层面**：LT 模式下，你可以通过 `strace` 看到 `poll` 反复被唤醒，从而判断“我是不是没把 `timerfd` drain 干净”。在 ET 模式里，你更依赖程序内部的日志或额外的状态检查，因为内核不会再帮你“提醒第二遍”。对初学者来说，这让用 `strace` 直接分析行为变得不那么便利。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Timeout!\n"
     ]
    }
   ],
   "source": [
    "! cd ../run && ./01_reactor_test01"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 8.2 TimerQueue定时器\n",
    "\n",
    "有了前面的Reactor基础，我们可以给`EventLoop`加上定时器功能。传统的Reactor通过控制`select(2)`和`poll(2)`的等待时间来实现定时，而现在在Linux中有了`timerfd`，我们可以用和处理IO事件相同的方式来处理定时，代码的一致性更好。muduo中的backport.diff展示了传统方案。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## TimerQueue class\n",
    "\n",
    "muduo的定时器功能由三个class实现，`TimerId`、`Timer`、`TimerQueue`，用户只能看到第一个class，另外两个都是内部实现细节。`TimerId`和`Timer`的实现很简单，这里就不展示源码了。\n",
    "\n",
    "`TimerQueue`的接口很简单，只有两个函数`addTimer()`和`cancel()`。本节我们只实现`addTimer()`，`cancel()`的实现见p.328-8.11。`addTimer()`是供`EventLoop`使用的，`EventLoop`会把它封装为更好用的`runAt()`、`runAfter()`、`runEvery()`等函数。\n",
    "\n",
    "- [TimerQueue.h](./02_timerqueue/TimerQueue.h#L32)\n",
    "\n",
    "值得一提的是`TimerQueue`的数据结构的选择，`TimerQueue`需要高效地组织目前尚未到期的`Timer`，能快速地根据当前时间找到已经到期的`Timer`，也要能高效地添加和删除`Timer`。最简单的`TimerQueue`以按到期时间排序的线性表为数据结构，muduo最早也是用这种结构。这种结构的常用操作是线性查找，复杂度是`O(N)`。\n",
    "\n",
    "- 另一种常用做法是二叉堆组织优先队列（`libev`用的是更高效的`4-heap`），这种做法的复杂度降为`O(log N)`，但是C++标准库的`make_heap()`等函数不能高效地删除heap中间的某个元素，需要我们自己实现（令`Timer`记住自己在heap中的位置）。\n",
    "- 还有一种做法是使用二叉搜索树（例如`std::set`/`std::map`），把`Timer`按到期时间先后排序。操作的复杂度仍然是`O(log N)`，不过memory locality比heap要差一些，实际速度可能略慢。但是我们不能直接用`map<Timestamp, Timer*>`，因为这样无法处理两个`Timer`到期时间相同的情况。有两个解决方案，一是用`multimap`或`multiset`，二是设法区分key。muduo现在采用的是第二种做法，这样可以避免使用不常见的`multimap class`。具体来说，以`pair<Timestamp, Timer*>`为key，这样即便两个`Timer`的到期时间相同，它们的地址也必定不同。\n",
    "\n",
    "> `std::set` 使用红黑树维护元素的严格弱序\n",
    "\n",
    "以下是`TimerQueue`的数据成员，这个结构利用了现成的容器库，实现简单，容易验证其正确性，并且性能也不错。`TimerList`是`set`而非`map`，因为只有`key`没有`value`。`TimerQueue`使用了一个`Channel`来观察`timerfd`上的`readable`事件。注意`TimerQueue`的成员函数只能在其所属的`IO`线程调用，因此不必加锁。\n",
    "\n",
    "- [TimerQueue.h](./02_timerqueue/TimerQueue.h#L49)\n",
    "\n",
    "TimerQueue 的实现目前有一个不理想的地方，Timer 是用裸指针管理的，需要手动 delete。这里用 `shared_ptr` 似乎有点小题大做了。在 C++11 中，或许可以改进为 `unique_ptr`，避免手动管理资源。\n",
    "\n",
    "来看关键的 `getExpired()` 函数的实现，这个函数会从 `timers_` 中移除已到期的 Timer，并通过 `vector` 返回它们。编译器会实施 RVO 优化，不必太担心性能，必要时可以像 `EventLoop::activeChannels_` 那样复用 `vector`。注意其中哨兵值（sentry）的选取，sentry 让 `set::lower_bound()` 返回的是第一个未到期的 Timer 的迭代器，因此[断言](./02_timerqueue/TimerQueue.cc#L145)中是 ` < ` 而非 ` <= `。\n",
    "\n",
    "- [TimerQueue.cc](./02_timerqueue/TimerQueue.cc#L140)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "图8-2是 `TimerQueue`回调用户代码`onTimer()`的时序图。\n",
    "\n",
    "<img src=\"./images/8.2.png\" alt=\"8.2\" style=\"zoom:40%;\" />"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## EventLoop的改动\n",
    "\n",
    "EventLoop新增了几个方便用户使用的定时器接口，这几个函数都转而调用 `TimerQueue::addTimer()`。注意这几个 `EventLoop` 成员函数应该允许跨线程调用， 比方说我想在某个IO线程中执行超时回调。这就带来线程安全性方面的问题，muduo的解决办法不是加锁， 而是把对 `TimerQueue` 的操作转移到IO线程来进行，这会用到 § 8.3 介绍的 `EventLoop::runInLoop()` 函数。\n",
    "\n",
    "- [EventLoop](./02_timerqueue/EventLoop.cc#L78)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "pid = 80143, tid = 80143\n",
      "now 1761549176.739544\n",
      "msg 1761549176.739842 main\n",
      "msg 1761549177.763111 once1\n",
      "msg 1761549178.242542 once1.5\n",
      "msg 1761549178.761276 every2\n",
      "msg 1761549179.241450 once2.5\n",
      "msg 1761549179.759643 every3\n",
      "msg 1761549180.248517 once3.5\n",
      "msg 1761549180.774452 every2\n",
      "msg 1761549182.760616 every3\n",
      "msg 1761549182.775747 every2\n",
      "msg 1761549184.789153 every2\n",
      "msg 1761549185.761357 every3\n",
      "msg 1761549186.797280 every2\n",
      "msg 1761549188.762722 every3\n",
      "msg 1761549188.800368 every2\n",
      "msg 1761549190.802691 every2\n",
      "msg 1761549191.770321 every3\n",
      "msg 1761549192.802911 every2\n",
      "msg 1761549194.774968 every3\n",
      "msg 1761549194.805961 every2\n",
      "msg 1761549194.805992 main loop exits\n"
     ]
    }
   ],
   "source": [
    "! cd ../run && ./02_timerqueue"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI:\n",
    "\n",
    "| 组件         | 定位与线程归属                                               | 主要职责                                                     | 关键交互                                                     | 典型回调/接口                                                |\n",
    "| ------------ | ------------------------------------------------------------ | ------------------------------------------------------------ | ------------------------------------------------------------ | ------------------------------------------------------------ |\n",
    "| `EventLoop`  | 每个线程唯一的事件循环调度器；运行于所属线程                 | 轮询 I/O、分发到 `Channel`、执行定时器与队列任务、处理跨线程唤醒 | 拥有 `Poller` 与 `TimerQueue`；维护 `activeChannels_`；通过 `wakeupFd_`/`wakeupChannel_` 被其他线程唤醒 | `loop()`, `runInLoop()`, `queueInLoop()`, `runAfter()`, `runEvery()` |\n",
    "| `Poller`     | I/O 多路复用抽象（`epoll`/`poll`/`kqueue` 封装）；与 `EventLoop` 同线程 | 等待事件并返回就绪的 `Channel` 列表；管理 `fd` → `Channel*` 的注册 | 由 `EventLoop` 持有；根据 `Channel` 的 `enable/disable/update` 调整关注事件 | `poll(timeout, activeChannels)`, `updateChannel()`, `removeChannel()` |\n",
    "| `Channel`    | `fd` 的事件订阅者与回调分发器；不拥有 `fd`                   | 保存关心的事件与用户回调；在事件就绪时调用回调               | 通过 `EventLoop -> Poller` 完成注册/更新/移除；可用 `tie()`/`std::weak_ptr` 绑定生命周期 | `setReadCallback()`, `setWriteCallback()`, `enableReading()`, `disableAll()`, `handleEvent()` |\n",
    "| `TimerQueue` | 定时器管理器；通常用 `timerfd` 驱动（其自身也有一个 `Channel`） | 维护按到期时间排序的定时任务，到期在 `EventLoop` 线程执行回调；支持一次性与周期性 | 由 `EventLoop` 持有；`timerfd` 的可读事件由 `Poller` 上报，触发 `TimerQueue::handleRead()` | `addTimer()`, `cancel()`, `handleRead()`（内部），与 `EventLoop` 的 `runAfter()`/`runEvery()` 配合 |\n",
    "\n",
    "- 所有权：`EventLoop` 拥有 `Poller`、`TimerQueue`、`wakeupChannel_`；`Poller` 维护 `fd -> Channel*` 的映射但不拥有 `Channel`；`Channel` 不拥有 `fd`。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- [类关系（所有权/关联）](./images/mermaid.md#类关系（所有权/关联）), [简化版类关系（仅所有权）](./02_timerqueue/mermaid.md#简化版类关系（仅所有权）)\n",
    "\n",
    "定时器路径：`TimerQueue(timerfd)` -> `Poller` -> `EventLoop` -> `TimerQueue::handleRead` -> 你的回调\n",
    "- [定时器注册与触发（时序）](./02_timerqueue/mermaid.md#定时器注册与触发（时序）)\n",
    "\n",
    "I/O 事件路径：OS -> `Poller` -> `EventLoop` -> `Channel` -> 你的回调\n",
    "- [普通 I/O 事件分发（时序）](./02_timerqueue/mermaid.md#普通I/O事件分发（时序）)\n",
    "\n",
    "跨线程任务：其他线程 -> `queueInLoop` + `wakeupFd` -> `EventLoop` -> 执行你的任务"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# * 8.3 `EventLoop::runInLoop()`函数\n",
    "\n",
    "`EventLoop` 有一个非常有用的功能：在它的 IO 线程内执行某个用户任务回调，即 `EventLoop::runInLoop(const Functor& cb)`，其中 Functor 是 `boost::function<void()>`。如果用户在当前 IO 线程调用这个函数，回调会同步进行；如果用户在其他线程调用 `runInLoop()`，`cb` 会被加入队列，IO `线程会被唤醒来调用这个 Functor。\n",
    "\n",
    "- [EventLoop::runInLoop()](./03_runInLoop/EventLoop.cc#L100)\n",
    "\n",
    "有了这个功能，我们就能轻易地在线程间调配任务，比如说把 `TimerQueue` 的成员函数调用移到其 IO 线程，这样可以在不用锁的情况下保证线程安全性。\n",
    "\n",
    "由于 IO 线程平时阻塞在事件循环 `EventLoop::loop()` 的 `poll(2)` 调用中，为了让 IO 线程能立刻执行用户回调，我们需要设法唤醒它。传统的办法是用 `pipe(2)`，IO 线程始终监视此管道的 `readable` 事件，在需要唤醒的时候，其他线程往管道里写一个字节，这样 IO 线程就从 `IO multiplexing` 阻塞调用中返回。（原理类似 `HTTP long polling`。）现在 Linux 有了 `eventfd(2)`，可以更高效地唤醒，因为它不必管理缓冲区。以下是 EventLoop 新增的成员。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "! diff -u ./02_timerqueue/EventLoop.h ./03_runInLoop/EventLoop.h "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```shell\n",
    " private:\n",
    "   void abortNotInLoopThread();\n",
    "+  void handleRead();  // waked up\n",
    "+  void doPendingFunctors();\n",
    " \n",
    "   typedef std::vector<Channel*> ChannelList;\n",
    " \n",
    "   bool looping_; /* atomic */\n",
    "   bool quit_; /* atomic */\n",
    "+  bool callingPendingFunctors_; /* atomic */\n",
    "   const pid_t threadId_;\n",
    "-  muduo::Timestamp pollReturnTime_;\n",
    "+  muduo::Timestamp pollReturnTime_; \n",
    "   boost::scoped_ptr<Poller> poller_;\n",
    "   boost::scoped_ptr<TimerQueue> timerQueue_;\n",
    "+  int wakeupFd_;\n",
    "+  // unlike in TimerQueue, which is an internal class,\n",
    "+  // we don't expose Channel to client.\n",
    "+  boost::scoped_ptr<Channel> wakeupChannel_;\n",
    "   ChannelList activeChannels_;\n",
    "+  muduo::MutexLock mutex_;\n",
    "+  std::vector<Functor> pendingFunctors_; // @GuardedBy mutex_\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`wakeupChannel_` 用于处理 `wakeupFd_` 上的 `readable` 事件，将事件分发至 `handleRead()` 函数。其中只有 `pendingFunctors_` 暴露给了其他线程，因此用 `mutex` 保护。\n",
    "\n",
    "`queueInLoop()` 的实现很简单，将 `cb` 放入队列，并在必要时唤醒 IO 线程。\n",
    "\n",
    "- [queueInLoop()](./03_runInLoop/EventLoop.cc#L112)\n",
    "\n",
    "“必要时”有两种情况，如果调用 `queueInLoop()` 的线程不是 IO 线程，那么唤醒是必须的；如果在 IO 线程调用 `queueInLoop()`，而此时正在调用 `pending functor`，那么也必须唤醒。换句话说，只有在 IO 线程的事件回调中调用 `queueInLoop()` 才无须 `wakeup()`。看了下面 `doPendingFunctors()` 的调用时间点，想必读者就能明白为什么。\n",
    "\n",
    "p.287-8.1.1 的事件循环 `EventLoop::loop()` 中需要增加一行代码，执行 `pendingFunctors_` 中的任务回调。\n",
    "\n",
    "- [EventLoop::loop()](./03_runInLoop/EventLoop.cc#L84)\n",
    "\n",
    "`EventLoop::doPendingFunctors()` 不是简单地在临界区内依次调用 Functor，而是把回调列表 `swap()` 到局部变量 `functors` 中，这样一方面减少了临界区的长度（意味着不会阻塞其他线程调用 `queueInLoop()`），另一方面也避免了死锁（因为 Functor 可能再调用 `queueInLoop()`）。\n",
    "\n",
    "- [EventLoop::doPendingFunctors()](./03_runInLoop/EventLoop.cc#L176)\n",
    "\n",
    "由于 `doPendingFunctors()` 调用的 Functor 可能再调用 `queueInLoop(cb)`，这时 `queueInLoop()` 就必须 `wakeup()`，否则这些新加的 `cb` 就不能被及时调用了。muduo 这里没有反复执行 `doPendingFunctors()` 直到 `pendingFunctors_` 为空，这是有意的，否则 IO 线程有可能陷入死循环，无法处理 IO 事件。\n",
    "\n",
    "剩下的事情就简单了，在 `EventLoop::quit()` 中增加几行代码，在必要时唤醒 IO 线程，让它及时终止循环。思考：为什么在 IO 线程调用 `quit()` 就不必 `wakeup()`？\n",
    "\n",
    "- [EventLoop::quit()](./03_runInLoop/EventLoop.cc#L91)\n",
    "\n",
    "`EventLoop::wakeup()` 和 `EventLoop::handleRead()` 分别对 `wakeupFd_` 写入数据和读出数据，代码从略。注意 muduo 不是在 `EventLoop::handleRead()` 中执行 `doPendingFunctors()`，理由见 http://blog.csdn.net/solstice/article/details/6171831#comments。\n",
    "\n",
    "`s03/test5.cc` 是单线程程序，测试了 `runInLoop()` 和 `queueInLoop()` 等新函数。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI:\n",
    "\n",
    "背后的约定\n",
    "- 每个 `EventLoop` 绑定一个线程（即 IO 线程），`loop()` 必须在这个线程里运行。\n",
    "- 其他线程如果要操作 `EventLoop`，需要通过 `runInLoop()`/`queueInLoop()` 把任务投递回 IO 线程，或者在提供的线程安全接口里调用。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI: TODO: 临界区\n",
    "\n",
    "`pendingFunctors_`作用流程如下：\n",
    "1. **其他线程调用 `queueInLoop()`**\n",
    "   - 用互斥锁保护，将回调函数（Functor）压入 `pendingFunctors_`。\n",
    "   - 若调用线程不是事件循环线程，或者当前事件循环正处理其他 Functor，会调用 `wakeup()` 通过 `eventfd` 将 `poll()` 唤醒。\n",
    "2. **事件循环线程在 `doPendingFunctors()` 中执行这些 Functor**\n",
    "   - 先锁住互斥量，把 `pendingFunctors_` 与局部向量 `functors` 交换，释放锁。\n",
    "   - 在锁外逐个执行任务，避免长时间持锁和潜在死锁。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI: TODO: 思考：为什么在 IO 线程调用 `quit()` 就不必 `wakeup()`？\n",
    "\n",
    "在 IO 线程内调用 `EventLoop::quit()` 时不需要 `wakeup()`，原因是：此时事件循环线程已经在执行回调或刚从 `poll()` 返回，并且处于“活跃状态”。`quit()` 只需把 `quit_` 设为 `true`，接下来循环在当前或下一轮检查到 `quit_` 为真，就会跳出 `while (!quit_)`，无需额外唤醒。\n",
    "\n",
    "相反，如果其他线程调用 `quit()`，IO 线程可能正阻塞在 `poll()`，最多会等到超时（比如 10 秒）才返回，不能及时退出；这时需要写入 `wakeupFd_` 唤醒 `poll()`，让它立刻检测到 `quit_` 并结束循环。因此，“不必 wakeup()”只成立于在 IO 线程上下文调用 `quit()` 的场景。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> IA: `runInLoop() / queueInLoop()` 提供同步与异步执行回调的功能:\n",
    "\n",
    "`runInLoop()` 和 `queueInLoop()` 都是 `EventLoop` 提供的跨线程任务调度接口，但语义不同：\n",
    "\n",
    "**`runInLoop(const Functor& cb)` —— “可能同步执行”**\n",
    "\n",
    "- **当前线程正是 `EventLoop` 所属线程**\n",
    "   直接执行 `cb()`，所以看起来是“同步”的。\n",
    "- **调用线程不是所属线程**\n",
    "   将回调丢到 `pendingFunctors_` 中，并唤醒事件循环线程，最终在 `doPendingFunctors()` 内执行，因而表现为异步。\n",
    "\n",
    "*用法建议*： <br/>\n",
    "当你希望“如果我已经在 IO 线程里，就立即执行；否则就让 IO 线程帮我执行”时，用 `runInLoop()`。\n",
    "\n",
    "\n",
    "**`queueInLoop(const Functor& cb)` —— “一定异步排队”**\n",
    "\n",
    "- 不管调用线程是不是事件循环线程，都会把 `cb` 放入 `pendingFunctors_` 队列。\n",
    "- 如果调用方不是事件循环线程，或者此刻事件循环线程正在执行其它 Functor（`callingPendingFunctors_ == true`），就会调用 `wakeup()` 通过 `eventfd` 唤醒 `poll()`。\n",
    "\n",
    "*用法建议*： <br/>\n",
    "当你明确希望“把任务排队，交给事件循环线程在下一轮执行”（避免当前栈继续执行）时，使用 `queueInLoop()`。例如：\n",
    "- 在事件回调中延后执行一些收尾操作，避免递归/重入。\n",
    "- 任何出于时序或锁顺序原因，需要明确异步化的场景。\n",
    "\n",
    "\n",
    "**与 `doPendingFunctors()` 的配合**\n",
    "\n",
    "- `pendingFunctors_` 受互斥锁保护，`doPendingFunctors()` 会在每次 `poll()` 返回后把待执行的 Functor 取出，并在事件循环线程中逐一调用。\n",
    "- 先 `swap` 到局部变量再执行，可以缩短持锁时间，避免 Functor 执行过程中再 `queueInLoop()` 引发死锁。\n",
    "\n",
    "\n",
    "**小结**\n",
    "\n",
    "| 函数          | 同线程调用效果 | 跨线程调用效果 | 典型场景                           |\n",
    "| ------------- | -------------- | -------------- | ---------------------------------- |\n",
    "| `runInLoop`   | 立即执行       | 排队异步执行   | 既要跨线程安全，又想尽可能同步执行 |\n",
    "| `queueInLoop` | 排队异步执行   | 排队异步执行   | 必须交给下一轮执行、避免重入/递归  |\n",
    "\n",
    "理解这两个接口的语义，有助于你在多线程环境里安全地将任务派发到 `EventLoop` 所属的 IO 线程。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI: TODO: 不是在 `EventLoop::handleRead()` 中执行 `doPendingFunctors()`的理由?\n",
    "\n",
    "1. 唤醒机制与执行职责的分离\n",
    "\n",
    "- `handleRead()` 是 `wakeupChannel_` 对应的读回调，仅在别的线程调用 `EventLoop::wakeup()` 写入 `eventfd` 时才会触发。\n",
    "- 如果把 `doPendingFunctors()` 放到 `handleRead()`，那么任何想被及时执行的回调（即便由 IO 线程自身注册）都需要走一遍 `wakeup()` → `poll()` → `handleRead()`，多出了 `write/poll/read` 三次系统调用。\n",
    "- 当前设计让 `handleRead()` 只负责清掉 `eventfd` 的可读状态，`doPendingFunctors()` 放在主循环统一调用，唤醒过程简单、开销低；任务真正执行的责任交给 `loop()` 主体来处理。\n",
    "\n",
    "2. 减少 IO 线程内部调度的额外开销\n",
    "\n",
    "- 当回调在 IO 线程内部通过 `queueInLoop()` 注册时（比如在某个 Channel 的事件处理函数里再次派发任务），主循环遍历完本轮所有 `Channel` 后，立刻就会执行 `doPendingFunctors()`，无须额外 wakeup。\n",
    "- 如果 `doPendingFunctors()` 放在 `handleRead()` 里，即使在 IO 线程里注册回调，也要人为地写 wakeup，从而导致多余的系统调用。当前设计避免了这部分开销。\n",
    "\n",
    "3. 保持事件处理顺序一致，避免重入\n",
    "\n",
    "- `handleRead()` 是众多活跃 `Channel` 回调之一。如果在处理其中一个（`wakeupChannel_`）时就执行 `pendingFunctors`，会打断 “先处理完本轮所有 IO 事件，再处理跨线程任务” 的既定顺序。\n",
    "- `pendingFunctors` 里的 Functor 可能对其他 Channel 做增删，甚至再次注册回调。如果在遍历 `activeChannels_` 的过程中间插入这些操作，容易引发迭代器失效、重入或状态混乱。\n",
    "- 把 `doPendingFunctors()` 固定在每次 `poll()` 返回、所有 `handleEvent()` 结束之后执行，流程更清晰，顺序可控。\n",
    "\n",
    "4. 配合 `callingPendingFunctors_` 等状态管理\n",
    "\n",
    "- `doPendingFunctors()` 会设置并清除 `callingPendingFunctors_`，与 `queueInLoop()` 的逻辑互相配合。\n",
    "- 如果在 `handleRead()`（单个 Channel 回调）里插入 `doPendingFunctors()`，这些状态的管理会变复杂，甚至影响 `queueInLoop()` 是否需要唤醒的判断，增加写错的概率。\n",
    "\n",
    "\n",
    "**小结**\n",
    "- `handleRead()` 仅负责处理 `eventfd` 的唤醒信号；\n",
    "- `doPendingFunctors()` 固定在 `EventLoop::loop()` 每一轮的末尾执行；\n",
    "- 这套设计让唤醒机制轻量，避免 IO 线程自投递任务时的额外系统调用，也确保事件处理顺序与线程安全性。\n",
    "\n",
    "因此，Muduo 不把 `doPendingFunctors()` 放到 `EventLoop::handleRead()` 中，是经过深思熟虑的架构选择。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## * 提高`TimerQueue`的线程安全性\n",
    "\n",
    "前面提到 `TimerQueue::addTimer()` 只能在 IO 线程调用，因此 `EventLoop::runAfter()` 系列函数不是线程安全的。下面这段代码在 § 8.2 中会 crash，因为它在非 IO 线程调用了 `EventLoop::runAfter()`。\n",
    "\n",
    "- [crash.cpp](./03_runInLoop/crash/crash.cpp)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "main(): pid = 177975, flag = 0\n",
      "run1(): pid = 177975, flag = 1\n",
      "run2(): pid = 177975, flag = 1\n",
      "run3(): pid = 177975, flag = 2\n",
      "run4(): pid = 177975, flag = 3\n",
      "main(): pid = 177975, flag = 3\n"
     ]
    }
   ],
   "source": [
    "! ../run/03_runInLoop"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "20251027 12:00:35.378340Z 178026 FATAL EventLoop::abortNotInLoopThread - EventLoop 0x7ffc90765f80 was created in threadId_ = 178025, current thread id = 178026 - EventLoop.cc:151\n",
      "Aborted\n"
     ]
    }
   ],
   "source": [
    "! ../run/03_runInLoop_crash"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "借助 `EventLoop::runInLoop()`，我们可以很容易地将 `TimerQueue::addTimer()` 做成线程安全的，而且无需用锁。办法是让 `addTimer()` 调用 `runInLoop()`，把实际工作转移到 IO 线程来做。先新增一个 `addTimerInLoop()` 成员函数：\n",
    "\n",
    "然后把`addTimer()`拆成两部分，拆分后的`addTimer()`只负责转发，a`ddTimerInLoop()`完成修改定时器列表的工作\n",
    "\n",
    "- [TimerQueue::addTimerInLoop](./03_runInLoop/TimerQueue.cc#L122)\n",
    "\n",
    "这样无论在哪个线程调用`addTimer()`都是安全的了，上一页的代码也能正常运行。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "! ../run/03_runInLoop_crash"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## `EventLoopThread class`\n",
    "\n",
    "IO 线程不一定是主线程，我们可以在任何一个线程创建并运行 `EventLoop`。一个程序也可以有不止一个 IO 线程，我们可以按优先级将不同的 `socket` 分给不同的 IO 线程，避免优先级反转。为了方便将来使用，我们定义 `EventLoopThread` class，这正是 `one loop per thread` 的本意。\n",
    "\n",
    "`EventLoopThread` 会启动自己的线程，并在其中运行 `EventLoop::loop()`。其中关键的 `startLoop()` 函数定义如下，这个函数会返回新线程中 `EventLoop` 对象的地址，因此用条件变量等待线程的创建与运行。\n",
    "\n",
    "- [EventLoopThread::startLoop()](./03_runInLoop/EventLoopThread.cc#L33)\n",
    "\n",
    "线程主函数在`stack`上定义`EventLoop`对象，然后将其地址赋值给`loop_`成员变量，最后`notify()`条件变量，唤醒`startLoop()`\n",
    "\n",
    "- [EventLoopThread::threadFunc()](./03_runInLoop/EventLoopThread.cc#L49)\n",
    "\n",
    "由于 EventLoop 的生命周期与线程主函数的作用域相同，因此在 threadFunc() 退出之后这个指针就失效了。好在服务器程序一般不要求能安全地退出（§ 9.2.2），这应该不是什么大问题。\n",
    "\n",
    "`s03/test6.cc` 测试了 `EventLoopThread` 的功能，也测试了跨线程调用 `EventLoop::runInLoop()` 和 `EventLoop::runAfter()`，代码从略。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "main(): pid = 185201, tid = 185201\n",
      "exit main().\n",
      "runInThread(): pid = 185201, tid = 185202\n"
     ]
    }
   ],
   "source": [
    "! ../run/03_eventloopthread"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI: 悬空指针\n",
    "\n",
    "- `EventLoop loop;` 放在线程函数栈上是为了保证事件循环对象与线程同生共死，并严格绑定线程上下文，这是 Muduo 的原始设计。\n",
    "- 使用时要清楚其生命周期：`startLoop()` 返回的指针仅在 `EventLoopThread` 存活期间有效"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI: 共享状态（`loop_` 指针）和条件变量（`cond_`）\n",
    "\n",
    "在 `EventLoopThread` 的实现里，`startLoop()` 与 `threadFunc()` 之间需要通过共享状态（`loop_` 指针）和条件变量（`cond_`）进行同步。这就决定了哪里必须加锁，原理是典型的生产者-消费者（或线程间同步）模式：一个线程等待条件满足，另一个线程在条件满足后发信号"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI:\n",
    "\n",
    "- [定时器注册与触发（时序）](./03_runInLoop/mermaid.md#定时器注册与触发（时序）,基于8.3)\n",
    "- [`runInLoop()`（调用线程即`EventLoop`线程）](./03_runInLoop/mermaid.md#`runInLoop()`（调用线程即`EventLoop`线程）)\n",
    "- [`runInLoop()`（调用线程 ≠ `EventLoop`线程）](./03_runInLoop/mermaid.md#`runInLoop()`（调用线程≠EventLoop线程）)\n",
    "- [`queueInLoop()`（统一异步入队）](./03_runInLoop/mermaid.md#`queueInLoop()`（统一异步入队）)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 8.4 实现TCP网络库\n",
    "\n",
    "到目前为止，Reactor事件处理框架已初具规模，从本节开始我们用它逐步实现一个非阻塞TCP网络编程库。从`poll(2)`返回到再次调用`poll(2)`阻塞称为一次事件循环。图8-3值得印在脑中，它有助于理解一次循环中各种回调发生的顺序。\n",
    "\n",
    "<img src=\"./images/8.3.png\" alt=\"8.3\" style=\"zoom:40%;\" />\n",
    "\n",
    "传统的 Reactor 实现一般会把 `timers` 做成循环中单独的一步，而 muduo 把它和 IO handlers 等同视之，这是使用 `timerfd` 的附带效应。将来有必要时也可以在调用 IO handlers 之前或之后处理 `timers`。\n",
    "\n",
    "后面几节的内容安排如下：\n",
    "- 8.4 介绍 `Acceptor class`，用于 `accept(2)` 新连接。\n",
    "- 8.5 介绍 `TcpServer`，处理新建 `TcpConnection`。\n",
    "- 8.6 处理 `TcpConnection` 断开连接。\n",
    "- 8.7 介绍 `Buffer class` 并用它读取数据。\n",
    "- 8.8 介绍如何无阻塞发送数据。\n",
    "- 8.9 完善 `TcpConnection`，处理 `SIGPIPE`、`TCP keep alive` 等。\n",
    "\n",
    "至此，单线程TCP服务端网络编程已经基本成型，大部分muduo示例都可以运行。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Acceptor class\n",
    "\n",
    "先定义`Acceptor class`，用于`accept(2)`新TCP连接，并通过回调通知使用者。它是内部`class`，供`TcpServer`使用，生命期由后者控制。`Acceptor`的接口如下\n",
    "\n",
    "- [class Acceptor](./04_tcp/tcp/Acceptor.h#L40)\n",
    "\n",
    "`Acceptor`的成员数据包括`Socket`、`Channel`等。 其中`Socket`是一个RAII handle，封装了`socket`文件描述符的生命周期。A`cceptor`的`socket`是`listening socket`，即`server socket`。`Channel`用于观察此socket上的`readable`事件，并回调`Acceptor::handleRead()`，后者会调用`accept(2)`来接受新连接，并回调用户`callback`。\n",
    "\n",
    "Acceptor的构造函数和 `Acceptor::listen()` 成员函数执行创建TCP服务端的传统步骤，即调用 `socket(2)`、`bind(2)`、`listen(2)` 等 Sockets API，其中任何一个步骤出错都会造成程序终止，因此这里看不到错误处理。\n",
    "> 通常原因是端口被占用。这时让程序异常退出更好，因为能触发监控系统报警，而不是假装正常运行\n",
    "\n",
    "- [Acceptor::Acceptor](./04_tcp/tcp/Acceptor.cc#L18)\n",
    "\n",
    "Acceptor的接口中用到了 `InetAddress` class，这是对 `struct sockaddr_in` 的简单封装，能自动转换字节序，代码从略。`InetAddress` 具备值语义，是可以拷贝的。\n",
    "\n",
    "Acceptor的构造函数用到 `createNonblockingOrDie()` 来创建非阻塞的 `socket`，现在的 Linux 可以一步完成(§ 4.11)，代码如下。\n",
    "\n",
    "- [sockets::createNonblockingOrDie](./04_tcp/tcp/SocketsOps.cc#L53)\n",
    "\n",
    "`Acceptor::listen()` 的最后一步让 `acceptChannel_` 在 `socket` 可读的时候调用 `Acceptor::handleRead()`，后者会接受（`accept(2)`）并回调 `newConnectionCallback_`。这里直接把 `socket fd` 传给 `callback`，这种传递 `int` 句柄的做法不够理想，在 `C++11` 中可以先创建 `Socket` 对象，再用移动语义把 `Socket` 对象 `std::move()` 给回调函数，确保资源的安全释放。\n",
    "\n",
    "- [Acceptor::handleRead](./04_tcp/tcp/Acceptor.cc#L38)\n",
    "\n",
    "注意这里的实现没有考虑文件描述符耗尽的情况，muduo 的处理办法见 § [7.7](../CH07/CH07.ipynb#7.7_限制服务器的最大并发连接数)。还有一个改进措施，在拿到大于或等于 `0` 的 `connfd` 之后，非阻塞地 `poll(2)` 一下，看看 `fd` 是否可读写。正常情况下 `poll(2)` 会返回 `writable`，表明 `connfd` 可用。如果 `poll(2)` 返回错误，表明 `connfd` 有问题，应该立刻关闭连接。\n",
    "\n",
    "`Acceptor::handleRead()` 的策略很简单，每次 `accept(2)` 一个 `socket`。另外还有两种实现策略，一是每次循环 `accept(2)`，直至没有新的连接到达；二是每次尝试 `accept(2)` N 个新连接，N 的值一般是 10。后面这两种做法适合短连接服务，而 muduo 是为长连接服务优化的，因此这里用了最简单的办法。这三种策略的对比见论文[《(accept)able Strategies for Improving Web Server Performance》](http://static.usenix.org/event/usenix04/tech/general/brecht.html)。\n",
    "\n",
    "利用 Linux 新增的系统调用可以直接 `accept(2)` 一步得到非阻塞的 `socket`。\n",
    "\n",
    "- [sockets::accept](./04_tcp/tcp/SocketsOps.cc#L94)\n",
    "\n",
    "这里区分致命错误和暂时性错误，并区别对待。对于暂时错误，例如 `EAGAIN`、`EINTR`、`EMFILE`、`ECONNABORTED` 等等，处理办法是忽略这次错误。对于致命错误，例如 `ENFILE`、`ENOMEM` 等等，处理办法是终止程序，对于未知错误也照此办理。\n",
    "\n",
    "下面写个小程序来试验`Acceptor`的功能，它在`9981`端口侦听新连接，连接到达后向它发送一个字符串，随即断开连接。\n",
    "\n",
    "- [test7.cc](./04_tcp/test7.cc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "! ../run/04_tcp\n",
    "main(): pid = 221455\n",
    "newConnection(): accepted a new connection from 127.0.0.1:36206"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "! curl 127.0.0.1:9981 hello world\n",
    "curl: (1) Received HTTP/0.9 when not allowed\n",
    "curl: (6) Could not resolve host: hello\n",
    "curl: (6) Could not resolve host: world"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 练习1：把[test7.cc](./04_tcp/test7.cc)改写为daytime服务器。: [daytime.cc](./04_tcp/daytime.cc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "! ../run/04_daytime"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Trying 127.0.0.1...\n",
      "Connected to 127.0.0.1.\n",
      "Escape character is '^]'.\n",
      "2025-10-27 23:35:30\n",
      "Connection closed by foreign host.\n"
     ]
    }
   ],
   "source": [
    "! telnet 127.0.0.1 9981"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 练习2：把[test7.cc](./04_tcp/test7.cc)扩充为同时侦听两个port，每个port发送不同的字符串。: [dualport.cc](./04_tcp/dualport.cc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "! ../run/04_dualport"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Trying 127.0.0.1...\n",
      "Connected to 127.0.0.1.\n",
      "Escape character is '^]'.\n",
      "Hello from port 9981!\n",
      "Connection closed by foreign host.\n"
     ]
    }
   ],
   "source": [
    "! telnet 127.0.0.1 9981"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Trying 127.0.0.1...\n",
      "Connected to 127.0.0.1.\n",
      "Escape character is '^]'.\n",
      "Greetings from port 9982!\n",
      "Connection closed by foreign host.\n"
     ]
    }
   ],
   "source": [
    "! telnet 127.0.0.1 9982"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 8.5 TcpServer接受新连接\n",
    "\n",
    "本节会介绍 `TcpServer` 并初步实现 `TcpConnection`，本节只处理连接的建立，下一节处理连接的断开，再往后依次处理读取数据和发送数据。\n",
    "\n",
    "`TcpServer` 新建连接的相关函数调用顺序见图 8-4（有的函数名是简写，省略了 `poll(2)` 调用）。其中 `Channel::handleEvent()` 的触发条件是 `listening socket` 可读，表明有新连接到达。`TcpServer` 会为新连接创建对应的 `TcpConnection` 对象。\n",
    "\n",
    "<img src=\"./images/8.4.png\" alt=\"8.4\" style=\"zoom:40%;\" />"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## `TcpServer class`\n",
    "\n",
    "`TcpServer` class 的功能是管理 `accept(2)` 获得的 `TcpConnection`。`TcpServer` 是供用户直接使用的，生命周期由用户控制。`TcpServer` 的接口如下，用户只需要设置好 `callback`，再调用 `start()` 即可。\n",
    "\n",
    "- [class TcpServer](./05_tcpserver/tcp/TcpServer.h#L26)\n",
    "\n",
    "`TcpServer` 内部使用 `Acceptor` 来获得新连接的 `fd`。它保存用户提供的 `ConnectionCallback` 和 `MessageCallback`，在新建 `TcpConnection` 的时候会原样传给后者。`TcpServer` 持有目前存活的 `TcpConnection` 的 `shared_ptr`（定义为 `TcpConnectionPtr`），因为 `TcpConnection` 对象的生命周期是模糊的，用户也可以持有 `TcpConnectionPtr`。\n",
    "\n",
    "- [class TcpServer](./05_tcpserver/tcp/TcpServer.h#L47)\n",
    "\n",
    "每个 `TcpConnection` 对象有一个名字，这个名字是由其所属的 `TcpServer` 在创建 `TcpConnection` 对象时生成，名字是 `ConnectionMap` 的 key。\n",
    "\n",
    "在新连接到达时，`Acceptor` 会回调 `newConnection()`，后者会创建 `TcpConnection` 对象 `conn`，把它加入 `ConnectionMap`，设置好 `callback`，再调用 `conn->connectEstablished()`，其中会调用用户提供的 `ConnectionCallback`。代码如下。\n",
    "\n",
    "- [TcpServer::newConnection](./05_tcpserver/tcp/TcpServer.cc#L50)\n",
    "\n",
    "- 练习：给 `TcpServer` 的构造函数增加 `string` 参数，用于初始化 `name_` 成员变量。\n",
    "\n",
    "注意 muduo 尽量让依赖是单向的，`TcpServer` 会用到 `Acceptor`，但 `Acceptor` 并不知道 `TcpServer` 的存在。`TcpServer` 会创建 `TcpConnection`，但 `TcpConnection` 并不知道 `TcpServer` 的存在。另外 L64 可以考虑改用 `make_shared()` 以节约一次 `new`。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## TcpConnection class\n",
    "\n",
    "`TcpConnection` class 可谓是 muduo 最核心也是最复杂的 class，它的头文件和源文件一共有 450 多行，是 muduo 最大的 class。本章会用 5 节的篇幅来逐渐完善它。\n",
    "\n",
    "`TcpConnection` 是 muduo 里唯一默认使用 `shared_ptr` 来管理的 class，也是唯一继承 `enable_shared_from_this` 的 class，这源于其模糊的生命周期，原因见 § 4.7。\n",
    "\n",
    "- [TcpConnectionPtr](./05_tcpserver/Callbacks.h)\n",
    "- [class TcpConnection](./05_tcpserver/tcp/TcpConnection.h#L33)\n",
    "\n",
    "本节的 `TcpConnection` 没有可供用户使用的函数，因此接口从略，以下是其数据成员。目前 `TcpConnection` 的状态只有两个，`kConnecting` 和 `kConnected`，后面几节会逐渐丰富其状态。`TcpConnection` 使用 `Channel` 来获得 `socket` 上的 IO 事件，它会自己处理 `writable` 事件，而把 `readable` 事件通过 `MessageCallback` 传达给客户端。`TcpConnection` 拥有 TCP `socket`，它的析构函数会 `close(fd)`（在 `Socket` 的析构函数中发生）。\n",
    "\n",
    "- [class TcpConnection](./05_tcpserver/tcp/TcpConnection.h#L61)\n",
    "\n",
    "注意 `TcpConnection` 表示的是“一次 TCP 连接”，它是不可再生的，一旦连接断开，这个 `TcpConnection` 对象就没用了。另外 `TcpConnection` 没有发起连接的功能，其构造函数的参数是已经建立好连接的 `socket fd`（无论是 `TcpServer` 被动接受还是 `TcpClient` 主动发起），因此其初始状态是 `kConnecting`。\n",
    "\n",
    "本节的 `MessageCallback` 定义很原始，没有使用 `Buffer` class，而只是把 `(const char* buf, int len)` 传给用户，这种接口用起来无疑是很不方便的。\n",
    "\n",
    "- [class TcpConnection](./05_tcpserver/tcp/TcpConnection.cc#L57)\n",
    "\n",
    "本节的 `TcpConnection` 只处理了建立连接，没有处理断开连接（例如 `handleRead()` 中 `read(2)` 返回 `0`），接收数据的功能很简陋，也不支持发送数据，这些都会逐步得到完善。\n",
    "\n",
    "[`test8.cc`](./05_tcpserver/test8.cc) 试验了目前实现的功能，它实际上是个 discard 服务。但目前它永远不会关闭 `socket`，即永远走不到 [`else` 分支](./05_tcpserver/test8.cc#L15)，在遇到对方断开连接的时候会陷入 busy loop。§ 8.6 会处理连接的断开。\n",
    "\n",
    "- [test8.cc](./05_tcpserver/test8.cc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# * 8.6 TcpConnection断开连接\n",
    "\n",
    "muduo 只有一种关闭连接的方式：被动关闭（p.191）。即对方先关闭连接，本地 `read(2)` 返回 0，触发关闭逻辑。将来如果有必要也可以给 `TcpConnection` 新增 `forceClose()` 成员函数，用于主动关闭连接，实现很简单，调用 `handleClose()` 即可。函数调用的流程见图 8-5，其中的 “X” 表示 `TcpConnection` 通常会在此时析构。\n",
    "\n",
    "<img src=\"./images/8.5.png\" alt=\"8.5\" style=\"zoom:40%;\" />\n",
    "\n",
    "一般来讲数据的删除比新建要复杂，TCP连接也不例外。关闭连接的流程看上去有点“绕”​，根本原因是p.274-7.15.3讲的对象生命期管理的需要。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## `Channel`的改动\n",
    "\n",
    "`Channel class`新增了`CloseCallback`事件回调，并且断言（`assert()`）在事件处理期间本`Channel`对象不会析构，即不会发生p.274-7.15.3讲的出错情况。\n",
    "\n",
    "- [Channel::~Channel()](./06_tcpconnection/reactor/Channel.cc#L34)\n",
    "- [closeCallback_](./06_tcpconnection/reactor/Channel.cc#L51)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## `TcpConnection`的改动\n",
    "\n",
    "`TcpConnection class`也新增了`CloseCallback`事件回调，但是这个回调是给`TcpServer`和`TcpClient`用的，用于通知它们移除所持有的`TcpConnectionPtr`，这不是给普通用户用的，普通用户继续使用`ConnectionCallback`。\n",
    "\n",
    "- [setCloseCallback](./06_tcpconnection/tcp/TcpConnection.h#L57)\n",
    "\n",
    "`TcpConnection`把另外几个`handle*()`事件处理函数也补上了，`handleWrite()`暂时为空。`Channel`的`CloseCallback`会调用`TcpConnection::handleClose()`，依此类推。\n",
    "\n",
    "- [handleClose](./06_tcpconnection/tcp/TcpConnection.h#L70)\n",
    "\n",
    "`TcpConnection::handleRead()`会检查`read(2)`的返回值，根据返回值分别调用`messageCallback_`、`handleClose()`、`handleError()`。\n",
    "\n",
    "- [TcpConnection::handleRead](./06_tcpconnection/tcp/TcpConnection.cc#L79)\n",
    "\n",
    "`TcpConnection::handleClose()`的主要功能是调用`closeCallback_`，这个回调绑定到`TcpServer::removeConnection()`。\n",
    "\n",
    "- [TcpConnection::handleClose](./06_tcpconnection/tcp/TcpConnection.cc#L92)\n",
    "\n",
    "`TcpConnection::handleError()`并没有进一步的行动，只是在日志中输出错误消息，这不影响连接的正常关闭。\n",
    "\n",
    "- [TcpConnection::handleError](./06_tcpconnection/tcp/TcpConnection.cc#L103)\n",
    "\n",
    "`TcpConnection::connectDestroyed()`是`TcpConnection`析构前最后调用的一个成员函数，它通知用户连接已断开。其中的`L69`与上面的`L98`重复，这是因为在某些情况下可以不经由`handleClose()`而直接调用`connectDestroyed()`。\n",
    "\n",
    "- [TcpConnection::connectDestroyed](./06_tcpconnection/tcp/TcpConnection.cc#L64)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## `TcpServer`的改动\n",
    "\n",
    "`TcpServer`向`TcpConnection`注册`CloseCallback`，用于接收连接断开的消息。\n",
    "\n",
    "- [TcpServer::newConnection](./06_tcpconnection/tcp/TcpServer.cc#L68-L70)\n",
    "\n",
    "通常`TcpServer`的生命周期长于它建立的`TcpConnection`，因此不用担心`TcpServer`对象失效。在`muduo`中，`TcpServer`的析构函数会关闭连接，因此也是安全的。\n",
    "\n",
    "`TcpServer::removeConnection()`把`conn`从`ConnectionMap`中移除。这时`TcpConnection`已经是命悬一线：如果用户不持有`TcpConnectionPtr`的话，`conn`的引用计数已降到`1`。注意这里一定要用`EventLoop::queueInLoop()`，否则就会出现`p.274`讲的对象生命周期管理问题。另外注意这里用`boost::bind`让`TcpConnection`的生命周期长到调用`connectDestroyed()`的时刻。\n",
    "\n",
    "- [TcpServer::removeConnection](./06_tcpconnection/tcp/TcpServer.cc#L73-L83)\n",
    "\n",
    "思考并验证：如果用户不持有`TcpConnectionPtr`，那么`TcpConnection`对象究竟在什么时候析构？\n",
    "\n",
    "有兴趣的读者可以单步跟踪连接断开的流程，[`test8.cc`](./06_tcpconnection/test8.cc)不会陷入`busy loop`。目前的做法不是最简洁的，但是可以几乎原封不动地用到多线程`TcpServer`中(§ 8.10)。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## `EventLoop`和`Poller`的改动\n",
    "\n",
    "本节`TcpConnection`不再是只生不灭，因此要求`EventLoop`也提供`unregister`功能。`EventLoop`新增了`removeChannel()`成员函数，它会调用`Poller::removeChannel()`，后者定义如下，复杂度为`O(log N)`。\n",
    "\n",
    "- [Poller::removeChannel](./06_tcpconnection/reactor/Poller.cc#L95)\n",
    "\n",
    "注意其中从数组`pollfds_`中删除元素是`O(1)`复杂度，办法是将待删除的元素与最后一个元素交换，再`pollfds_.pop_back()`。这需要相应地修改`p.286-8.1.2`的代码：\n",
    "\n",
    "- [Poller::updateChannel](./06_tcpconnection/reactor/Poller.cc#L86-L91)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# * 8.7 `Buffer`读取数据\n",
    "\n",
    "`Buffer`是非阻塞TCP网络编程必不可少的东西(§ 7.4)，本节介绍用`Buffer`来处理数据输入，下一节介绍数据输出。`Buffer`是另一个具有值语义的对象。\n",
    "\n",
    "首先修改`Callbacks.h`中`MessageCallback`的定义，现在的参数和`muduo`一样，是`Buffer*`和`Timestamp`，不再是原始的`(const char* buf, int len)`。\n",
    "\n",
    "- [MessageCallback](./07_buffer/Callbacks.h#L30)\n",
    "\n",
    "其中`Timestamp`是`poll(2)`返回的时刻，即消息到达的时刻，这个时刻早于读到数据的时刻（`read(2)`调用或返回）。因此如果要比较准确地测量程序处理消息的内部延迟，应该以此时刻为起点，否则测出来的结果偏小，特别是处理并发连接时效果更明显。（*为什么？*）为此我们需要修改`Channel`中`ReadEventCallback`的原型，改动如下。`EventLoop::loop()`也需要有相应的改动，此处从略。\n",
    "\n",
    "- [ReadEventCallback](./07_buffer/reactor/Channel.h#L30)\n",
    "\n",
    "[test3.cc](./07_buffer/test8.cc#L30)试验了以上改动："
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI:\n",
    "\n",
    "为什么以 `poll` 返回时间为准？\n",
    "\n",
    "- **`poll()` 返回时刻早于 `read()` 调用**\n",
    "   当 `poll()` 检测到某个 fd 可读，它会立即返回；随后 `EventLoop` 才会依次遍历 `activeChannels` 并调用对应的 `handleEvent()`，在这之后才真正执行 `read()`。因此 `pollReturnTime_` 记录的是“内核通知用户态”这一刻，而 `read()` 完成的时间已经晚了若干个函数调用和队列调度的开销。\n",
    "- **并发连接场景下差异放大**\n",
    "   如果有多个连接同时变为可读，`poll()` 会把它们一次性返回。`EventLoop` 会顺序处理 `activeChannels` 列表：排在后面的连接要等前面的连接处理完（包括其回调逻辑）才能真正 `read()`。因此：\n",
    "  - 以 `read()` 完成时间为起点，会忽略掉该连接在 `activeChannels` 队列中“排队等待”的时间。\n",
    "  - 当并发连接数增多、每个连接的回调处理时间较长时，这段排队时间可能远大于真正读取数据的时间，从而导致测得的“内部延迟”偏小甚至全失真。\n",
    "- **`pollReturnTime_` 更贴近真实到达时间**\n",
    "   `poll()` 能够立即感知数据在内核缓冲区就绪的时刻，这是应用层可拿到的最早的时间点。将它作为“消息到达”时间，再与业务逻辑完成的时间相减，能够更准确地衡量应用层处理延迟。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## `TcpConnection`使用`Buffer`作为输入缓冲\n",
    "\n",
    "先给`TcpConnection`添加`inputBuffer_`成员变量。\n",
    "\n",
    "- [inputBuffer_](./07_buffer/tcp/TcpConnection.h#L85)\n",
    "\n",
    "然后修改`TcpConnection::handleRead()`成员函数，使用`Buffer`来读取数据。\n",
    "\n",
    "- [TcpConnection::handleRead](./07_buffer/tcp/TcpConnection.cc#L77-L80)\n",
    "\n",
    "修改test8.cc以试验本次改动后的新功能。\n",
    "\n",
    "- [test3.cc](./07_buffer/test8.cc#L30)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## `Buffer::readFd()`\n",
    "\n",
    "我在p.208-7.4.3提到Buffer读取数据时兼顾了内存使用量和效率，其实现如下。\n",
    "\n",
    "- [Buffer::readFd()](./07_buffer/tcp/Buffer.cc#L18-L38)\n",
    "\n",
    "这个实现有几点值得一提。一是使用了 scatter/gather IO，并且一部分缓冲区取自 stack，这样输入缓冲区足够大，通常一次 `readv(2)` 调用就能取完全部数据。由于输入缓冲区足够大，也节省了一次 `ioctl(socketFd, FIONREAD, &length)` 系统调用，不必事先知道有多少数据可读而提前预留（`reserve()`）`Buffer` 的 `capacity()`，可以在一次读取之后再将 `extrabuf` 中的数据 `append()` 给 `Buffer`。\n",
    "\n",
    "二是 `Buffer::readFd()` 只调用一次 `read(2)`，而没有反复调用 `read(2)` 直到其返回 `EAGAIN`。首先，这么做是正确的，因为 `muduo` 采用 level trigger，这么做不会丢失数据或消息。其次，对追求低延迟的程序来说，这么做是高效的，因为每次读数据只需要一次系统调用。再次，这样做照顾了多个连接的公平性，不会因为某个连接上数据量过大而影响其他连接处理消息。\n",
    "\n",
    "假如 `muduo` 采用 edge trigger，那么每次 `handleRead()` 至少调用两次 `read(2)`，平均起来比 level trigger 多一次系统调用，edge trigger 不见得更高效。\n",
    "\n",
    "将来的一个改进措施是：如果 `n == writable + sizeof extrabuf`，就再读一次。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# * 8.8 `TcpConnection`发送数据\n",
    "\n",
    "发送数据比接收数据更难，因为发送数据是主动的，接收读取数据是被动的。这也是本章先介绍`TcpServer`后介绍`TcpClient`的原因。到目前为止，我们只用到了`Channel`的`ReadCallback`：\n",
    "\n",
    "- `TimerQueue`用它来读`timerfd(2)`。\n",
    "- `EventLoop`用它来读`eventfd(2)`。\n",
    "- `TcpServer/Acceptor`用它来读`listening socket`。\n",
    "- `TcpConnection`用它来读普通`TCP socket`。\n",
    "\n",
    "本节会动用其`WriteCallback`，由于`muduo`采用`level trigger`，因此我们只在需要时才关注`writable`事件，否则就会造成`busy loop`。`s08/Channel.h`的改动如下：\n",
    "\n",
    "- [Channel](./08_tcpconnection/reactor/Channel.h#L50-L55)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI: **busy loop（忙等待循环）**\n",
    "\n",
    "当 muduo 使用的是 **Level Trigger（电平触发）** 模式时，内核会在 **套接字当前处于可写状态** 时反复通知你“这个 fd 可写”。而对于大多数 TCP 连接来说，只要发送缓冲区里还有空间，套接字几乎一直被视为可写。\n",
    "\n",
    "如果你在事件循环中 **始终监听 `writable` 事件**（即从一开始就把“关注可写”这个标志加到 epoll 里），那么：\n",
    "\n",
    "1. epoll 会在每次轮询时都返回“可写”。\n",
    "2. 事件循环看到“可写”就去处理，但发现暂时没数据要写，于是无事可做。\n",
    "3. 下一次 epoll 仍然返回“可写”，事件循环又被唤醒，却还是没有真正的写操作。\n",
    "4. 如此往复，事件循环就会不断在“可写事件”上空转，占满 CPU —— 这就是所谓的 **busy loop（忙等待循环）**。\n",
    "\n",
    "为了避免这种情况，muduo 的做法是：\n",
    "\n",
    "- **只有当发送缓冲区里有待发送的数据时，才去关注 `writable` 事件**；\n",
    "- 一旦数据写完，立即 **取消对 `writable` 事件的关注**。\n",
    "\n",
    "这样，只有在真正需要写数据、且之前因为内核缓冲区满而写不出去的时候，才会收到“可写”的通知；当没有数据需要写时，就不会被可写事件不断唤醒，从而避免 CPU 空转。\n",
    "\n",
    "> AI: **套接字当前处于可写状态** \n",
    "\n",
    "在 TCP 连接中，“套接字当前处于可写状态”意味着：**内核的发送缓冲区里还有足够的剩余空间，可以接受你提交的更多数据**。当你调用 `write`/`send` 等系统调用时，数据会先从用户态缓冲区拷贝到内核态的发送缓冲区。如果该缓冲区没被填满，写操作通常会立刻成功返回，这就是“可写”的表现。\n",
    "\n",
    "相反，如果发送缓冲区已经被占满（例如对方接收速度慢，导致内核缓冲区积压），套接字就变成“不可写”：写操作要么被阻塞（阻塞模式），要么返回 `EAGAIN`/`EWOULDBLOCK`（非阻塞模式），表示需要等到稍后缓冲区有空位。\n",
    "\n",
    "因此，“套接字当前处于可写状态”本质上描述的是内核发送缓冲区的可用性，它决定了你是否可以立即将更多数据推送出去。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "TcpConnection的接口中增加了`send()`和`shutdown()`两个函数，这两个函数都可以跨线程调用。为了简单起见，本章只提供一种`send()`重载。\n",
    "\n",
    "- [TcpConnection::send](./08_tcpconnection/tcp/TcpConnection.h#L51-L56)\n",
    "\n",
    "TcpConnection的状态增加到了4个，和目前muduo的实现一致。\n",
    "\n",
    "- [`enum StateE{kConnecting, kConnected, kDisconnecting, kDisconnected,};`](./08_tcpconnection/tcp/TcpConnection.h#L73-L74)\n",
    "\n",
    "其内部实现增加了两个`*InLoop`成员函数，对应前面的两个新接口函数，并使用`Buffer`作为输出缓冲区。\n",
    "\n",
    "- [TcpConnection](./08_tcpconnection/tcp/TcpConnection.h#L80-L82)\n",
    "- [Buffer outputBuffer_](./08_tcpconnection/tcp/TcpConnection.h#L95-L96)\n",
    "\n",
    "TcpConnection有一个非常简单的状态图（见图8-6）。\n",
    "\n",
    "<img src=\"./images/8.6.png\" alt=\"8.6\" style=\"zoom:40%;\" />\n",
    "\n",
    "TcpConnection在关闭连接的过程中与其他操作（读写事件）的交互比较复杂，尚需完备的单元测试来验证各种时序下的正确性。必要时可能要新增状态。\n",
    "\n",
    "`shutdown()`是线程安全的，它会把实际工作放到`shutdownInLoop()`中来做，后者保证在IO线程调用。如果当前没有正在写入，则关闭写入端(p.191-为什么 `TcpConnection::shutdown()` 没有直接关闭 TCP 连接)。代码注释给出了两个值得改进的地方。\n",
    "- [shutdown()](./08_tcpconnection/tcp/TcpConnection.cc#L94-L114)\n",
    "\n",
    "由于新增了`kDisconnecting`状态，`TcpConnection::connectDestroyed()`和`TcpConnection::handleClose()`中的`assert()`也需要相应的修改，代码从略。\n",
    "\n",
    "- [TcpConnection::connectDestroyed()](./08_tcpconnection/tcp/TcpConnection.cc#L127-L128)\n",
    "- [TcpConnection::handleClose()](./08_tcpconnection/tcp/TcpConnection.cc#L179-L180)\n",
    "\n",
    "`send()`也是一样的，如果在非IO线程调用，它会把`message`复制一份，传给IO线程中的`sendInLoop()`来发送。这么做或许有轻微的效率损失，但是线程安全性很容易验证，我认为还是利大于弊。如果真的在乎这点性能，不如让程序只在IO线程调用`send()`。另外在C++11中可以使用移动语义，避免内存拷贝的开销。\n",
    "\n",
    "- [TcpConnection::send()](./08_tcpconnection/tcp/TcpConnection.cc#L54-L65)\n",
    "\n",
    "`sendInLoop()`会先尝试直接发送数据，如果一次发送完毕就不会启用`WriteCallback`；如果只发送了部分数据，则把剩余的数据放入`outputBuffer_`，并开始关注`writable`事件，以后在`handlerWrite()`中发送剩余的数据。如果当前`outputBuffer_`已经有待发送的数据，那么就不能先尝试发送了，因为这会造成数据乱序。\n",
    "\n",
    "- [TcpConnection::sendInLoop()](./08_tcpconnection/tcp/TcpConnection.cc#L66-L93)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI: `TcpConnection::sendInLoop()`\n",
    "\n",
    "下面从整体流程、关键判断和相关机制三个角度，讲解 `TcpConnection::sendInLoop()` 的工作原理，以及它为何要这么设计。\n",
    "\n",
    "------\n",
    "\n",
    "**1. 整体流程概述**\n",
    "\n",
    "`sendInLoop()` 负责在事件循环线程中发送数据。它的核心步骤如下：\n",
    "\n",
    "1. **确保在 I/O 线程执行**：`loop_->assertInLoopThread()` 会断言当前调用线程就是事件循环线程，防止并发问题。\n",
    "2. **尝试直接写 socket**：如果此前没有监听 `writable` 事件，且输出缓冲区 `outputBuffer_` 为空，就直接调用 `::write` 向 socket 写数据。\n",
    "3. **处理写操作结果**：\n",
    "   - 如果一次写完（`nwrote == message.size()`），直接返回，不再触发后续写回调。\n",
    "   - 如果只写出一部分（或写失败但错误是 `EWOULDBLOCK`），会把剩余数据放入 `outputBuffer_`，并开启对 `writable` 事件的关注。\n",
    "4. **注册写事件**：当存在未写完的数据时，通过 `channel_->enableWriting()` 让事件循环在 socket 可写时回调 `handlerWrite()`，继续发送缓冲区数据。\n",
    "\n",
    "\n",
    "\n",
    "**2. 关键判断逻辑详解**\n",
    "\n",
    "`if (!channel_->isWriting() && outputBuffer_.readableBytes() == 0)`\n",
    "\n",
    "- **含义**：只有在当前没有监听写事件，且输出缓冲为空时，才“冒险”尝试直接写。\n",
    "- **原因**：如果缓冲里本来就有数据，说明之前就未写完；此时直接写新数据，会打乱先后顺序。保证先到的数据先发送，是网络协议栈的基本要求。\n",
    "\n",
    "`nwrote = ::write(...)`\n",
    "\n",
    "- **直接写 socket** 意味着同步尝试。结果有三种：\n",
    "  1. **全部写完**：无需额外处理。\n",
    "  2. **部分写完**：说明 socket 发送缓冲已满，需要后续异步继续写。\n",
    "  3. **写失败**：\n",
    "     - 如果错误是 `EWOULDBLOCK`（资源暂时不可用），说明 socket 现在不可写，和“部分写完”一样处理即可。\n",
    "     - 如果是其他错误，则记录日志，有可能要处理连接异常（代码中只是记录 `LOG_SYSERR`）。\n",
    "\n",
    "`outputBuffer_.append(...)` & `channel_->enableWriting()`\n",
    "\n",
    "- **追加剩余数据**：将没写完的部分添加到 `outputBuffer_` 中。\n",
    "- **监听写事件**：调用 `channel_->enableWriting()` 后，当 socket 变得可写（事件循环检测到 `EPOLLOUT` 等）时，就会触发 `handlerWrite()`，把缓冲里的数据继续发送出去。\n",
    "\n",
    "\n",
    "\n",
    "**3. 为什么要避免乱序？**\n",
    "\n",
    "> “如果当前 `outputBuffer_` 已经有待发送的数据，那么就不能先尝试发送了，因为这会造成数据乱序。”\n",
    "\n",
    "原因是这样的：\n",
    "\n",
    "- `outputBuffer_` 中的数据比当前 `message` 更早提交给 `sendInLoop()`，但还没能写出去。\n",
    "- 若此时直接 `write` 新消息，而把旧消息滞留在缓冲区，就会让新数据比旧数据更早上网线，违背了应用层对数据顺序的期望。\n",
    "- 因此，一旦缓冲中已有待发送的数据，就必须先把缓冲区的旧数据通过 `handlerWrite()` 发完，再发送新数据。\n",
    "\n",
    "\n",
    "\n",
    "**4. 与 `WriteCallback` 的关系**\n",
    "\n",
    "- 在 Muduo 中，`WriteCallback` 通常跟 `channel_->handleEvent()` 的 `EPOLLOUT` 事件关联。\n",
    "- 如果一次性写完，就没有必要关注 `EPOLLOUT`，因此不会触发 `WriteCallback`。\n",
    "- 一旦有剩余数据，就会注册写事件。待 kernel 通知 socket 可写时，`Channel` 会回调 `TcpConnection::handleWrite()`，此时 `handleWrite()` 会从 `outputBuffer_` 取数据继续发，直到写完或再次写满。\n",
    "\n",
    "\n",
    "\n",
    "**5. 小结**\n",
    "\n",
    "`sendInLoop()` 的设计目标，就是**尽量减少延迟**（能直接写就直接写），同时**保证数据顺序**和**异步发送的可靠性**。它利用输出缓冲与写事件配合，实现了以下几点：\n",
    "\n",
    "- **低延迟**：在缓冲为空时尝试直接写，避免多余的事件轮训。\n",
    "- **顺序保证**：一旦缓冲中有旧数据，就先处理旧数据，防止乱序。\n",
    "- **适配非阻塞 IO**：当 socket 暂时不可写时，剩余数据会保存在缓冲里，等待下一次可写事件再发送。\n",
    "- **错误处理**：及时记录异常，并通过外部（比如连接关闭逻辑）进一步处理。\n",
    "\n",
    "这段代码是 Muduo 中经典的“高性能、非阻塞 TCP 写”模式，非常值得借鉴。\n",
    "\n",
    "------"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "当`socket`变得可写时，`Channel`会调用`TcpConnection::handleWrite()`，这里我们继续发送`outputBuffer_`中的数据。一旦发送完毕，立刻停止观察`writable`事件(L160)，避免`busy loop`。另外如果这时[连接正在关闭](./08_tcpconnection/tcp/TcpConnection.cc#L161)，则调用`shutdownInLoop()`，继续执行关闭过程。这里不需要处理错误，因为一旦发生错误，`handleRead()`会读到0字节，继而关闭连接。\n",
    "\n",
    "- [`TcpConnection::handleWrite()`](./08_tcpconnection/tcp/TcpConnection.cc#L150-L174)\n",
    "\n",
    "注意`sendInLoop()`和`handleWrite()`都只调用了一次`write(2)`而不会反复调用直至它返回`EAGAIN`，原因是如果第一次`write(2)`没有能够发送完全部数据的话，第二次调用`write(2)`几乎肯定会返回`EAGAIN`。读者可以很容易用下面的Python代码来验证这一点。因此muduo决定节省一次系统调用，这么做不影响程序的正确性，却能降低低延迟。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "python"
    }
   },
   "outputs": [],
   "source": [
    "#!/usr/bin/python\n",
    "import socket, sys\n",
    "\n",
    "sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)\n",
    "sock.connect(('remote_hostname', 9876))  # 这里最好连接到网络上的一台机器\n",
    "sock.setblocking(0)\n",
    "a = 'a' * int(sys.argv[1])  # 两条消息的长度由命令行给出，a 应该足够大\n",
    "b = 'b' * int(sys.argv[2])\n",
    "\n",
    "n1 = sock.send(a)  # 第一次发送\n",
    "n2 = 0\n",
    "try:\n",
    "    n2 = sock.send(b)  # 第二次发送，遇到 EAGAIN 会抛 socket.error 异常\n",
    "except socket.error as ex:\n",
    "    print ex  # socket.error: [Errno 11] Resource temporarily unavailable\n",
    "print n1\n",
    "print n2\n",
    "sock.close()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "一个改进措施：`TcpConnection`的输出缓冲区不必是连续的（`outputBuffer_`改成`ptr_vector<Buffer>`），`handleWrite()`可以用`writev(2)`来发送多块数据，这样或许能减少内存拷贝的次数，略微提高性能（但这种性能提高不一定能被外界感知）。\n",
    "\n",
    "在`level trigger`模式中，数据的发送比较麻烦，因为不能一直关注`writable`事件，不过数据的读取很简单。我认为理想的做法是对`readable`事件采用`level trigger`，对`writable`事件采用`edge trigger`，但是目前Linux不支持这种设定。\n",
    "\n",
    "`test9.cc`是 `echo server(§ 6.4.2)`，代码从略。`s08/test10.cc`试验`TcpConnection::send()`的功能，它和前面的Python示例相近，都是通过命令行指定两条消息的大小，然后连续发送两条消息。通过选择不同的消息长度，可以试验不同的`code path`。\n",
    "\n",
    "- [test9.cc](./08_tcpconnection/test9.cc)\n",
    "- [test10.cc](./08_tcpconnection/test10.cc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 8.9 完善`TcpConnection`\n",
    "\n",
    "至此`TcpConnection`的主体功能接近完备，可以应付大部分muduo示例的需求了。本节补充几个小功能，让它成为可以实用的单线程非阻塞TCP网络库。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## SIGPIPE\n",
    "\n",
    "SIGPIPE的默认行为是终止进程，在命令行程序中这是合理的，但是在网络编程中，这意味着如果对方断开连接而本地继续写入的话，会造成服务进程意外退出。\n",
    "\n",
    "> 在命令行程序中，默认的 SIGPIPE 行为非常有用。例如查看日志中的前 10 条错误信息，可以用管道将命令串起来：`gunzip -c log.gz | grep ERROR | head`,  由于 head 关闭了管道的写入端，grep 会遇到 SIGPIPE 而终止，同理 `gunzip` 也就不需要解压缩整个巨大的日志文件。这也可能是 Unix 默认使用阻塞 IO 的历史原因之一。\n",
    "\n",
    "假如服务进程繁忙，没有及时处理对方断开连接的事件，就有可能出现在连接断开之后继续发送数据的情况。下面这个例子模拟了这种情况：\n",
    "\n",
    "- [test10.cc](./09_tcpconnection_opt/test10.cc#L17-L21)\n",
    "\n",
    "假设sleepSeconds是5秒，用`nc localhost 9981`创建连接之后立刻Ctrl-C断开客户端，服务进程过几秒就会退出。解决办法很简单，在程序开始的时候就忽略SIGPIPE，可以用C++全局对象做到这一点。\n",
    "\n",
    "- [`class IgnoreSigPipe`](./09_tcpconnection_opt/reactor/EventLoop.cc#L37-L45)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "! nc localhost 9981"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI:\n",
    "\n",
    "服务器往往在一个进程里维护大量连接，某个连接的断开是常态，不应该导致整个进程崩溃。尤其在非阻塞模式下，我们常常是“尽力写”：`send()` 返回 `EPIPE`、`write()` 返回 -1 并且同时触发 `SIGPIPE`。如果不屏蔽 `SIGPIPE`，服务进程就会被无情终止，造成在线服务中断。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## TCP No Delay和TCP keepalive\n",
    "\n",
    "TCP No Delay和TCP keepalive都是常用的TCP选项，前者的作用是禁用[Nagle算法](http://en.wikipedia.org/wiki/Nagle’s_algorithm)，避免连续发包出现延迟，这对编写低延迟网络服务很重要。后者的作用是定期探查TCP连接是否还存在。一般来说如果有应用层心跳的话，TCP keepalive不是必需的，但是一个通用的网络库应该暴露其接口。（本书不涉及`TCP_CORK`。）\n",
    "> 如果没有应用层心跳，而对方机器突然断电，那么本机不会收到TCP的`FIN`分节。在没有发送消息的情况下，这个“连接”可能一直保持下去。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI:\n",
    "\n",
    "------\n",
    "\n",
    "在编写通用的 TCP 网络库或服务框架时，两个常见的 socket 选项几乎必然会被考虑：`TCP_NODELAY` 与 `SO_KEEPALIVE`。它们分别解决“延迟”与“连接存活探测”这两类问题。\n",
    "\n",
    "\n",
    "**1. `TCP_NODELAY` —— 禁用 Nagle 算法**\n",
    "\n",
    "**Nagle 算法**在早期网络环境中用于提升网络带宽利用率：它会缓冲小包，等到收到 ACK 或凑够一定大小后再发，以减少碎片化的包数量。但在低延迟场景（比如 IM、在线游戏、RPC）里，这种“攒包”行为反而增加了响应延迟。\n",
    "\n",
    "`TCP_NODELAY` 选项用于禁用 Nagle，让应用层 `write()` 一次就立即发送一个 TCP 段。\n",
    "\n",
    "使用建议\n",
    "\n",
    "- **适合场景**：对“单次发送就希望立即到达”很敏感的应用（实时交互、快速响应 RPC 等）。\n",
    "- **注意事项**：禁用后小包会频繁发送，带宽占用和包数可能上升，对网络较差或设备繁忙的环境要权衡。\n",
    "- **示例代码**：\n",
    "  \n",
    "```cpp\n",
    "int flag = 1;\n",
    "::setsockopt(sockfd, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(flag));\n",
    "```\n",
    "\n",
    "- **默认状态**：通常默认启用 Nagle，即 `TCP_NODELAY` 关闭，需要显式设置。\n",
    "\n",
    "\n",
    "**2. TCP Keepalive —— 内核级连接探测**\n",
    "\n",
    "`SO_KEEPALIVE` 是一个 TCP 层的“存活探测”机制：如果在一段时间内没有数据交互，内核会周期性地发送探测包以判断连接是否仍然存在。\n",
    "\n",
    "适用场景\n",
    "\n",
    "- **无应用层心跳**：若应用层没有主动发送“心跳”或 ping，还希望自动发现“死链接”，可以考虑开启。\n",
    "- **防止半开连接**：比如远端断电或网络断开，本端没有发送数据时不会收到 `FIN`，连接就会长期挂着——Keepalive 能够在 N 分钟后发现异常并关闭。\n",
    "\n",
    "限制与代价\n",
    "\n",
    "- **触发延迟长**：各平台默认定时较长（Linux 默认约 2 小时；Windows 更长），需要手动调节才能满足实时需求。\n",
    "- **依赖内核策略**：不同操作系统参数和行为可能不同，在 Linux 上可以通过 `sysctl net.ipv4.tcp_keepalive_*` 调整全局默认值，也可以通过 `setsockopt` 设置 per-socket 数值。\n",
    "- **存在额外流量消耗**：保持机制会周期性发送探测包，虽然数量不多，但全局大量连接时需要考虑。\n",
    "\n",
    "示例代码（Linux）\n",
    "\n",
    "```cpp\n",
    "int enable = 1;\n",
    "::setsockopt(sockfd, SOL_SOCKET, SO_KEEPALIVE, &enable, sizeof(enable));\n",
    "\n",
    "// 设置更短的探测周期\n",
    "int idle = 60;      // 空闲 60 秒后开始探测\n",
    "int interval = 10;  // 探测间隔 10 秒\n",
    "int count = 5;      // 连续 5 次失败判定断开\n",
    "::setsockopt(sockfd, IPPROTO_TCP, TCP_KEEPIDLE, &idle, sizeof(idle));\n",
    "::setsockopt(sockfd, IPPROTO_TCP, TCP_KEEPINTVL, &interval, sizeof(interval));\n",
    "::setsockopt(sockfd, IPPROTO_TCP, TCP_KEEPCNT, &count, sizeof(count));\n",
    "```\n",
    "\n",
    "与应用层心跳的关系\n",
    "\n",
    "- **有心跳**：如果你已经在协议层设计了 ping/pong 或心跳包，一般不需要依赖 Keepalive。\n",
    "- **无心跳**：Keepalive 是一个备用手段，尤其是你无法简单改动应用协议时。\n",
    "\n",
    "**3. 为什么网络库应当暴露这两个开关？**\n",
    "\n",
    "- **场景差异大**：低延迟游戏服务器 vs. 日志采集服务，需求完全不同。\n",
    "- **默认值不一定合适**：标准库的默认策略可能与业务需求相悖，需要由业务来做最终决定。\n",
    "- **兼容性**：不同平台默认参数不同；暴露接口可以让调用者按需设置。\n",
    "\n",
    "常见的做法是在连接类（如 `TcpConnection`）上提供类似方法：\n",
    "\n",
    "```cpp\n",
    "void TcpConnection::setTcpNoDelay(bool on);\n",
    "void TcpConnection::setKeepAlive(bool on);\n",
    "```\n",
    "\n",
    "有些库还会进一步提供 Keepalive 参数配置接口。\n",
    "\n",
    "\n",
    "**4. 补充：`TCP_CORK`（了解即可）**\n",
    "\n",
    "虽然你提到“本书不涉及 `TCP_CORK`”，但简单区分一下有助于理解：\n",
    "\n",
    "- `TCP_NODELAY` 是禁止内核延迟发送。\n",
    "- `TCP_CORK` 则是相反的：尽量“攒包”，直到你明确告诉内核可以发送或者缓冲区满。适用于需要先拼出完整大包再发的场景（比如 HTTP 响应头和正文一次性发出）。\n",
    "\n",
    "`TCP_CORK` 与 `TCP_NODELAY` 互斥（在 Linux 上设置一个会自动禁用另一个），因此在设计 API 时要注意双方的语义冲突。\n",
    "\n",
    "\n",
    "**5. 总结**\n",
    "\n",
    "| 选项           | 目的                       | 适用场景                           | 代价/注意事项                                      |\n",
    "| -------------- | -------------------------- | ---------------------------------- | -------------------------------------------------- |\n",
    "| `TCP_NODELAY`  | 禁用 Nagle，降低发送延迟   | 低延迟交互、实时响应               | 会增加小包数量，可能略微降低吞吐效率               |\n",
    "| `SO_KEEPALIVE` | 探测“半开”连接（远端失联） | 没有应用层心跳、需要自动清理死连接 | 默认间隔很长，需要调节；对资源和网络有少量额外消耗 |\n",
    "\n",
    "对于通用网络库，通常会提供设置这些选项的接口，让上层业务按需选择是否启用、并调整参数。这也是 Muduo 等成熟网络库的惯用设计。\n",
    "\n",
    "------"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "以下是`TcpConnection::setTcpNoDelay()`的实现，涉及3个文件。\n",
    "\n",
    "- [TcpConnection::setTcpNoDelay](./09_tcpconnection_opt/tcp/TcpConnection.h#L56)\n",
    "- [Socket::setTcpNoDelay](./09_tcpconnection_opt/tcp/Socket.cc#L60-L67)\n",
    "\n",
    "[`TcpConnection::setKeepAlive()`](../muduo/muduo/net/Socket.cc#L121-L128)的实现与之类似，此处从略，可参考muduo源码。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## * `WriteCompleteCallback`和`HighWaterMarkCallback`\n",
    "\n",
    "非阻塞网络编程的发送数据比读取数据要困难得多：一方面是§ 8.8提到的“什么时候关注writable事件”的问题，这只带来编码方面的难度；另一方面是如果发送数据的速度高于对方接收数据的速度，会造成数据在本地内存中堆积，这带来设计及安全性方面的难度。muduo对此的解决办法是提供两个回调，有的网络库把它们称为“高水位回调”和“低水位回调”，muduo使用 `HighWaterMarkCallback` 和 `WriteCompleteCallback` 这两个名字。`WriteCompleteCallback` 很容易理解，如果发送缓冲区被清空，就调用它。`TcpConnection` 有两处可能触发此回调：\n",
    "\n",
    "- [`TcpConnection::sendInLoop`](./09_tcpconnection_opt/tcp/TcpConnection.cc#L76-L79)\n",
    "- [`TcpConnection::sendInLoop`](./09_tcpconnection_opt/tcp/TcpConnection.cc#L168-L172)\n",
    "\n",
    "`TcpConnection`和`TcpServer`也需要相应地暴露`WriteCompleteCallback`的接口，代码从略。\n",
    "\n",
    "[test11.cc](./09_tcpconnection_opt/test11.cc)是chargen服务（§ 7.1），用到了`WriteCompleteCallback`，代码从略。\n",
    "\n",
    "另外一个有用的`callback`是`HighWaterMarkCallback`，如果输出缓冲的长度超过用户指定的大小，就会触发回调（只在上升沿触发一次）。代码见muduo，此处从略。\n",
    "\n",
    "如果用非阻塞的方式写一个proxy，proxy有C和S两个连接（§ 7.13）。只考虑server发给client的数据流（反过来也是一样），为了防止server发过来的数据撑爆C的输出缓冲区，一种做法是在C的`HighWaterMarkCallback`中停止读取S的数据，而在C的`WriteCompleteCallback`中恢复读取S的数据。这就跟用粗水管往水桶里灌水，用细水管从水桶中取水一个道理，上下两个水龙头要轮流开合，类似PWM。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI: `WriteCompleteCallback` 与 `HighWaterMarkCallback` 的角色与协作:\n",
    "\n",
    "------\n",
    "\n",
    "在非阻塞网络编程中，**写数据的复杂度远高于读数据**。原因主要有两个：\n",
    "\n",
    "1. **何时监听写事件 (`writable`)**：需要避免不必要的写事件监听（详见之前的 `sendInLoop()` 分析）。\n",
    "2. **发送速度差异**：若本端发送速度远超对方接收速度，数据会不断堆积在本地缓冲区，造成内存膨胀甚至 DoS 漏洞。\n",
    "\n",
    "Muduo 的应对方式是提供两个关键回调：\n",
    "\n",
    "- `WriteCompleteCallback`\n",
    "- `HighWaterMarkCallback`\n",
    "\n",
    "它们分别在“发送缓冲清空”与“发送缓冲长度超限”时通知应用层，以便上层逻辑可以实施流控——这就是所谓的“低水位”和“高水位”策略（Muduo 没用低水位命名，但语义相当）。\n",
    "\n",
    "\n",
    "**1. `WriteCompleteCallback` —— 发送缓冲清空的信号**\n",
    "\n",
    "触发时机\n",
    "\n",
    "当 `TcpConnection` 的输出缓冲（通常是 `outputBuffer_`）被完全写出时触发。Muduo 内部有两个触发点：\n",
    "\n",
    "- `TcpConnection::sendInLoop()` 在发现一次写操作后缓冲清空（通常是 `channel_->disableWriting()` 后）触发。\n",
    "- `TcpConnection::handleWrite()` 处理完 `writable` 事件且缓冲耗尽时触发。\n",
    "\n",
    "作用\n",
    "\n",
    "- **恢复读取**：若你因高水位暂停了某个上游的读，可以在写缓冲清空后重新开启读。\n",
    "- **告知业务层**：例如通知应用“消息已彻底发送”，便于高层协议推进状态机或释放资源。\n",
    "- **节流逻辑**：对于 RPC、一问一答服务，可在此时发送下一条请求/响应。\n",
    "\n",
    "接口暴露\n",
    "\n",
    "`TcpConnection` 与 `TcpServer` 需要暴露设置 `WriteCompleteCallback` 的接口，例如：\n",
    "\n",
    "```cpp\n",
    "void TcpConnection::setWriteCompleteCallback(const WriteCompleteCallback& cb);\n",
    "```\n",
    "\n",
    "上层可以这样用：\n",
    "\n",
    "```cpp\n",
    "conn->setWriteCompleteCallback(\n",
    "  [](const TcpConnectionPtr& conn) {\n",
    "    // 输出缓冲清空，可能通知上层或重启上游读\n",
    "  }\n",
    ");\n",
    "```\n",
    "\n",
    "**2. `HighWaterMarkCallback` —— 输出缓冲上涨的警戒线**\n",
    "\n",
    "触发条件\n",
    "\n",
    "当 `outputBuffer_` 的长度首次超过预设阈值（High Water Mark）时触发，并且在缓冲长度回落到阈值以下之前不会重复触发（即“上升沿”触发）。\n",
    "\n",
    "Muduo 内的逻辑大致为：\n",
    "\n",
    "```cpp\n",
    "if (oldLen < highWaterMark_ && newLen >= highWaterMark_)\n",
    "  highWaterMarkCallback_(shared_from_this(), newLen);\n",
    "```\n",
    "\n",
    "作用\n",
    "\n",
    "- **防止缓冲无限增长**：当生产速度（上游写入）大于消费速度（底层 socket 打出去），应该立刻告知应用层减慢/暂停上游数据流。\n",
    "- **实现流控**：配合 `WriteCompleteCallback` 实现“水桶阀门”的开闭控制。\n",
    "\n",
    "典型用法\n",
    "\n",
    "```cpp\n",
    "conn->setHighWaterMarkCallback(\n",
    "  [](const TcpConnectionPtr& conn, size_t bytes) {\n",
    "    // bytes 表示 outputBuffer_ 当前长度\n",
    "    // 可以暂停从上游读取、打出警报等\n",
    "  },\n",
    "  highWaterThreshold\n",
    ");\n",
    "```\n",
    "\n",
    "------\n",
    "\n",
    "**3. 组合用例：TCP 代理（Proxy）中的流量平衡**\n",
    "\n",
    "一个代理连接由两段组成：\n",
    "\n",
    "- C（客户端连接，Client）\n",
    "- S（服务器连接，Server）\n",
    "\n",
    "在非阻塞模式下，假设我们关注 S→C 的数据流：\n",
    "\n",
    "1. **C 的输出缓冲被填充**（即客户端暂时收不动数据）：\n",
    "   - `HighWaterMarkCallback` 触发。\n",
    "   - 代理应 **暂停从 S 读取**，例如调用 `serverChannel->disableReading()`。\n",
    "   - 防止 S 继续推更多数据，挤爆 C 的缓冲。\n",
    "2. **C 输出缓冲清空**（客户端终于消费完）：\n",
    "   - `WriteCompleteCallback` 触发。\n",
    "   - 代理 **恢复从 S 读取**，例如 `serverChannel->enableReading()`。\n",
    "   - 相当于水桶（C 的输出缓冲）水位降回安全线，可以继续注水。\n",
    "\n",
    "这种策略就像“粗水管灌水、细水管放水”的 PWM（脉宽调制）原理：上游与下游两个阀门交替开关，避免溢出或干涸。\n",
    "\n",
    "伪代码示意\n",
    "\n",
    "```cpp\n",
    "// 当 C 侧 outputBuffer 超过阈值\n",
    "connC->setHighWaterMarkCallback(\n",
    "  [serverConn](const TcpConnectionPtr& conn, size_t bytes) {\n",
    "    serverConn->stopRead();  // 停止读取来自 S 的数据\n",
    "  },\n",
    "  highWaterThreshold\n",
    ");\n",
    "\n",
    "// 当 C 侧 outputBuffer 清空\n",
    "connC->setWriteCompleteCallback(\n",
    "  [serverConn](const TcpConnectionPtr& conn) {\n",
    "    serverConn->startRead(); // 恢复读取 S 的数据\n",
    "  }\n",
    ");\n",
    "```\n",
    "\n",
    "> 注意：真实实现中要考虑线程安全、生命周期和状态同步等细节。\n",
    "\n",
    "\n",
    "**4. 实现注意点**\n",
    "\n",
    "- **回调线程上下文**：Muduo 约定所有回调都在 I/O 线程（或对应 event loop 所在线程）执行。调用者若要在其他线程操控对象，需配合 `EventLoop::runInLoop()` 或 `queueInLoop()`。\n",
    "- **高水位阈值**：需根据业务和内存预算调整，不同连接可以有不同阈值。\n",
    "- **写缓冲大小监控**：即便有高水位回调，也建议服务端整体监控写缓冲占用，以防被恶意连接拖垮。\n",
    "- **与应用层协议结合**：例如在 `HighWaterMarkCallback` 中向上层报告“对端压力大”，让业务层减速。\n",
    "\n",
    "\n",
    "\n",
    "**5. 小结**\n",
    "\n",
    "| 回调                    | 触发条件               | 典型用途                                     |\n",
    "| ----------------------- | ---------------------- | -------------------------------------------- |\n",
    "| `WriteCompleteCallback` | 输出缓冲清空           | 通知业务发送已完成、重启上游读、推进协议状态 |\n",
    "| `HighWaterMarkCallback` | 输出缓冲第一次超过阈值 | 暂停上游读/写、施加流控、防止内存膨胀        |\n",
    "\n",
    "两者合用构成非阻塞写入的“流量闸门”：高水位提醒“停水”，低水位（即缓冲清空，虽然没有命名为“LowWater”但语义一致）提醒“开水”。在处理高速数据流时，它们能有效保护服务进程免于内存爆炸，同时保持较低延迟。\n",
    "\n",
    "------\n",
    "\n",
    "| 手段类别                    | 主要目标                  | 使用要点                                               |\n",
    "| --------------------------- | ------------------------- | ------------------------------------------------------ |\n",
    "| 高低水位/WriteComplete 回调 | 精细控制单连接的发送节奏  | Muduo 风格；对少量连接管理方便，可覆盖多数业务场景     |\n",
    "| 应用层额度/心跳机制         | 协议内的“主动背压”        | 需要协议支持；适合可控的上下游系统                     |\n",
    "| 限速/令牌桶                 | 平滑速率、限制峰值        | 在日志、监控、推送等高吞吐系统中常见                   |\n",
    "| 消息排队/丢弃               | 控制内存并做差异化处理    | 增加复杂度；适合允许降级或需要持久化的场景             |\n",
    "| 零拷贝传输                  | 减少用户态缓存参与        | 多见于代理、网关；仍需配合事件驱动的流控策略           |\n",
    "| OS 层参数                   | 辅助或补充策略            | `SO_SNDBUF`、`SO_SNDLOWAT` 等；平台差异要注意          |\n",
    "| 熔断/超时                   | 防止慢消费者拖累          | 对容忍度低的场景有效；需协商断开策略                   |\n",
    "| 事件循环节流/优先级调度     | 平衡 CPU/IO，控制回调粒度 | 在高并发情况下用于平滑负载或防止单连接占据过多事件资源 |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# * 8.10 多线程`TcpServer`\n",
    "\n",
    "本章的最后几节介绍三个主题：多线程TcpServer、TcpClient、epoll(4)，主题之间相互独立。\n",
    "\n",
    "本节介绍多线程`TcpServer`，用到了`EventLoopThreadPool class`。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## `EventLoopThreadPool`\n",
    "\n",
    "用 `one loop per thread` 的思想实现多线程 `TcpServer` 的关键步骤\n",
    "- 是在新建 `TcpConnection` 时从 `event loop pool` 里挑选一个 `loop` 给 `TcpConnection` 用。也就是说多线程 `TcpServer` 自己的 `EventLoop` 只用来接受新连接，而新连接会用其他 `EventLoop` 来执行 `IO`。（单线程 `TcpServer` 的 `EventLoop` 是与 `TcpConnection` 共享的。）`muduo` 的 `event loop pool` 由 `EventLoopThreadPool` class 表示，接口如下，实现从略。\n",
    "\n",
    "- [`class EventLoopThreadPool`](./10_TcpServer/reactor/EventLoopThreadPool.h#L27-L44)\n",
    "\n",
    "`TcpServer` 每次新建一个 `TcpConnection` 就会调用 `getNextLoop()` 来取得 `Event-Loop`，如果是单线程服务，每次返回的都是 `baseLoop_`，即 `TcpServer` 自己用的那个 `loop`。其中 `setThreadNum()` 的参数的意义见 `TcpServer` 代码注释。\n",
    "\n",
    "- [`class TcpServer`](./10_TcpServer/tcp/TcpServer.h#L32-L43)\n",
    "\n",
    "`TcpServer`只用增加一个成员函数和一个成员变量。\n",
    "\n",
    "- [`class TcpServer`](./10_TcpServer/tcp/TcpServer.h#L68-L72)\n",
    "- [`threadPool_`](./10_TcpServer/tcp/TcpServer.h#L78)\n",
    "\n",
    "多线程 `TcpServer` 的改动很简单，新建连接只改了 3 行代码。原来是把 `TcpServer` 自用的 `loop_` 传给 `TcpConnection`，现在是每次从 `EventLoopThreadPool` 取得 `ioLoop`。[`L81: ioLoop->runInLoop`](./10_TcpServer/tcp/TcpServer.cc#L81) 的作用是让 `TcpConnection` 的 `ConnectionCallback` 由 `ioLoop` 线程调用。\n",
    "\n",
    "- [`TcpServer::newConnection`](./10_TcpServer/tcp/TcpServer.cc#L72-L82)\n",
    "\n",
    "连接的销毁也不复杂，把原来的 `removeConnection()` 拆为两个函数，因为 `TcpConnection` 会在自己的 `ioLoop` 线程调用 `removeConnection()`，所以需要把它移到 `TcpServer` 的 `loop_` 线程（因为 `TcpServer` 是无锁的）。[`L98: ioLoop->queueInLoop`](./10_TcpServer/tcp/TcpServer.cc#L98) 再次把 `connectDestroyed()` 移到 `TcpConnection` 的 `ioLoop` 线程进行，是为了保证 `TcpConnection` 的 `ConnectionCallback` 始终在其 `ioLoop` 回调，方便客户端代码的编写。\n",
    "\n",
    "- [`removeConnection()`](./10_TcpServer/tcp/TcpServer.cc#L87)\n",
    "- [`removeConnectionInLoop()`](./10_TcpServer/tcp/TcpServer.cc#L893-L99)\n",
    "\n",
    "总而言之，`TcpServer` 和 `TcpConnection` 的代码都只处理单线程的情况（甚至都没有 `mutex` 成员），而我们借助 `EventLoop::runInLoop()` 并引入 `EventLoopThreadPool` 让多线程 `TcpServer` 的实现易如反掌。注意 `ioLoop` 和 `loop_` 间的线程切换都发生在连接建立和断开的时候，不影响正常业务的性能。\n",
    "\n",
    "`muduo` 目前采用最简单的 `round-robin` 算法来选取 `pool` 中的 `EventLoop`，不允许 `TcpConnection` 在运行中更换 `EventLoop`，这对长连接和短连接服务都是适用的，不易造成偏载。`muduo` 目前的设计是每个 `TcpServer` 有自己的 `EventLoopThreadPool`，多个 `TcpServer` 之间不共享 `EventLoopThreadPool`。将来如果有必要，也可以多个 `TcpServer` 共享 `EventLoopThreadPool`，比如说一个服务有多个等价的 `TCP` 端口，每个 `TcpServer` 负责一个端口，而来自这些端口的连接(s)共享一个 `EventLoopThreadPool`。\n",
    "\n",
    "另外一种可能的用法是一个 `EventLoop` `aLoop` 供两个 `TcpServer` 使用（`a` 和 `b`）。其中 `a` 是单线程服务，`aLoop` 既要 `accept(2)` 连接也要执行 `IO`；而 `b` 是多线程服务，有自己的 `EventLoopThreadPool`，只用 `aLoop` 来 `accept(2)` 连接。`aLoop` 上还可以运行几个 `TcpClient`。这些搭配都是可行的，这也正是 `EventLoop` 的灵活性所在，可以根据需要在多个线程间调配负载。本节更新了 `test8~test11`，均支持多线程。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI: `round-robin` 算法: \n",
    "\n",
    "“Round-robin”（轮询）本身并不是针对单个连接的背压或流量控制手段，它更像是一种**调度/分配策略**，主要应用在以下两个层面：\n",
    "\n",
    "1. **请求分发或负载均衡**\n",
    "   - 例如：多个后端节点时，轮询决定每个新连接/请求发给哪台机器。\n",
    "   - 目标是让整体负载大致均衡，而不是直接控制某条连接的发送速度或缓冲水位。\n",
    "2. **事件循环里的任务调度**\n",
    "   - 某些 Reactor 实现会按轮询顺序处理不同的事件源，以避免某个对象被连续处理太久。\n",
    "   - 这种做法可以间接防止某个连接“霸占”线程，也算是一种“公平性策略”，但它不等同于写缓冲的流控机制。\n",
    "\n",
    "**总结**：\n",
    "\n",
    "- Round-robin 更偏向“谁先得到服务”的公平性安排，不是直接控制“发送速率”或“缓冲上限”的工具。\n",
    "- 你可以将它**作为辅助策略**来防止某个连接独占资源，但真正意义上的背压仍需借助高水位/低水位、应用层限流、令牌桶等机制来实现。\n",
    "- 如果你希望在调度层面引入“轮询 + 背压”效果，可以将轮询调度与每个连接的水位阈值结合：轮到某连接时，只有它的缓冲水位安全才继续投喂，否则跳过。\n",
    "\n",
    "所以，「round-robin 算不算」要看你把它用在什么层面：它属于调度策略，能帮你实现公平性，但还需要搭配具体的流量控制手段来达到真正的背压效果。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# * 8.11 `Connector`\n",
    "\n",
    "主动发起连接比被动接受连接要复杂一些，一方面是错误处理麻烦，另一方面是要考虑重试。在非阻塞网络编程中，发起连接的基本方式是调用 `connect(2)`，当 `socket` 变得可写时表明连接建立完毕。当然这其中要处理各种类型的错误，因此我们把它封装为 `Connector` class。接口如下：\n",
    "\n",
    "- [`class Connector`](./11_connector/tcp/TcpServer.h#L25)\n",
    "\n",
    "`Connector` 只负责建立 `socket` 连接，不负责创建 `TcpConnection`，它的 `NewConnectionCallback` 回调的参数是 `socket` 文件描述符。以下是一个简单的测试（`s11/test12.cc`），它会反复尝试直至成功建立连接。\n",
    "\n",
    "`Connector` 的实现有几个难点：\n",
    "- `socket` 是一次性的，一旦出错（比如对方拒绝连接），就无法恢复，只能关闭重来。但 `Connector` 是可以反复使用的，因此每次尝试连接都要使用新的 `socket` 文件描述符和新的 `Channel` 对象。要留意 `Channel` 对象的生命周期管理，并防止 `socket` 文件描述符泄漏。\n",
    "- 错误代码与 `accept(2)` 不同，`EAGAIN` 是真的错误，表明本机 ephemeral port 暂时用完，要关闭 `socket` 再延期重试。“正在连接”的返回码是 `EINPROGRESS`。另外，即便出现 `socket` 可写，也不一定意味着连接已经成功建立，还需要用 `getsockopt(sockfd, SOL_SOCKET, SO_ERROR, …)` 再次确认一下。\n",
    "- 重试的间隔应该逐渐延长，例如 0.5s、1s、2s、4s，直至 30s，即 back-off。这会造成对象生命周期管理方面的困难，如果使用 `EventLoop::runAfter()` 定时而 `Connector` 在定时器到期之前析构了怎么办？本书的做法是在 `Connector` 的析构函数中注销定时器。\n",
    "- 要处理自连接（self-connection）。出现这种状况的原因如下。在发起连接的时候，TCP/IP 协议栈会先选择 source IP 和 source port，在没有显式调用 `bind(2)` 的情况下，source IP 由路由表确定，source port 由 TCP/IP 协议栈从 local port range 中选取尚未使用的 port（即 ephemeral port）。如果 destination IP 正好是本机，而 destination port 位于 local port range，且没有服务程序监听的话，ephemeral port 可能正好选中了 destination port，这就出现 `(source IP, source port) = (destination IP, destination port)` 的情况，即发生了自连接。处理办法是断开连接再重试，否则原本侦听 destination port 的服务进程也无法启动了。\n",
    "\n",
    "这里就不展示 `Connector` class 了，读者可以带着以上疑问去阅读 muduo 源码。\n",
    "\n",
    "- 练习 1：改写 [`test12.cc`](./11_connector/test12.cc)，通过命令行控制它发起 N 个并发连接，可用于测试 TCP 网络服务程序的并发性。注意这个练习可能没有想象中那么简单，如果同时发起 10000 个连接，那么某些 TCP SYN 分节可能丢包，而操作系统默认重发 SYN 的延时是 3 秒，我们无法直接控制。因此需要控制并发度，采用流水作业，尽量减少丢包。\n",
    "- 练习 2：验证自连接出现的情况。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## `TimerQueue::cancel()`\n",
    "\n",
    "§ 8.2 实现的 `TimerQueue` 不能注销定时器，本节补充这一功能。`TimerQueue::cancel()` 的一种简单实现是用 `shared_ptr` 来管理 `Timer` 对象，再将 `TimerId` 定义为 `weak_ptr<Timer>`，这样几乎不用我们做什么事情。在 C++11 中应该也足够高效，因为 `shared_ptr` 具备移动语义，可以做到引用计数值始终不变，没有原子操作的开销。但是用 `shared_ptr` 来管理 `Timer` 对象似乎显得有点小题大做，而且这种做法也有一个小小的缺点，如果用户一直持有 `TimerId`，会造成引用计数所占的内存无法释放，而本节展示的做法不会有这个问题。\n",
    "\n",
    "本节采用更传统的方式，保持现有的设计，让 `TimerId` 包含 `Timer*`。但这是不够的，因为无法区分地址相同的先后两个 `Timer` 对象。因此每个 `Timer` 对象有一个全局递增的序列号 `int64_t sequence_`（用原子计数器 `AtomicInt64` 生成），`TimerId` 同时保存 `Timer*` 和 `sequence_`，这样 `TimerQueue::cancel()` 就能根据 `TimerId` 找到需要注销的 `Timer` 对象。\n",
    "\n",
    "- [`s_numCreated_`](./11_connector/reactor/Timer.h#L52)\n",
    "\n",
    "`TimerQueue` 新增了 `cancel()` 接口函数，这个函数是线程安全的。\n",
    "\n",
    "- [`TimerQueue::cancel()`](./11_connector/reactor/TimerQueue.h#L47)\n",
    "\n",
    "`cancel()` 有对应的 `cancelInLoop()` 函数，因此 `TimerQueue` 不必用锁。`TimerQueue` 新增了几个数据成员，`activeTimers_` 保存的是目前有效的 `Timer` 的指针，并满足 invariant：`timers_.size() == activeTimers_.size()`，因为这两个容器保存的是相同的数据，只不过 `timers_` 是按到期时间排序，`activeTimers_` 是按对象地址排序。\n",
    "\n",
    "- [`activeTimers_`](./11_connector/reactor/TimerQueue.h#L75)\n",
    "\n",
    "由于 `TimerId` 不负责 `Timer` 的生命周期，其中保存的 `Timer*` 可能失效，因此不能直接 dereference，只有在 `activeTimers_` 中找到了 `Timer` 时才能提领。注销定时器的流程如下，照例用 `EventLoop::runInLoop()` 将调用转发到 IO 线程：\n",
    "\n",
    "- [`TimerQueue::cancel()`](./11_connector/reactor/TimerQueue.cc#L118-L120)\n",
    "\n",
    "上面这段代码中的 `cancelingTimers_` 和 `callingExpiredTimers_` 是为了应对“自注销”这种情况，即在定时器回调中注销当前定时器：\n",
    "\n",
    "- [test4.cc](./11_connector/test4.cc#L31)\n",
    "\n",
    "当运行到 [`g_loop->cancel(toCancel)`](./11_connector/test4.cc#L31) 的时候，`toCancel` 代表的 `Timer` 已经不在 `timers_` 和 `activeTimers_` 这两个容器中，而是 [`expired`](./11_connector/reactor/TimerQueue.cc#L159) 中（见 p.292 的 `getExpired()` 实现）。\n",
    "\n",
    "为了应对这种情况，`TimerQueue` 会记住在本次调用到期 `Timer` 期间有哪些 `cancel()` 请求，并且不再把已 `cancel()` 的 `Timer` 添加回 `timers_` 和 `activeTimers_` 当中。\n",
    "\n",
    "注意 `TimerQueue` 在执行 [`it->second->run();`](./11_connector/reactor/TimerQueue.cc#L167) 时没有检查 `Timer` 是否已撤销，这是因为 `TimerQueue::cancel()` 并不提供 strong guarantee。`TimerQueue::getExpired()` 和 `TimerQueue::insert()` 均增加了与 `activeTimers_` 有关的处理，此处从略。\n",
    "\n",
    "- [TimerQueue::reset](./11_connector/reactor/TimerQueue.cc#L203-L206)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# * 8.12 `TcpClient`\n",
    "\n",
    "有了 `Connector`，`TcpClient` 就不难实现了，它的代码与 `TcpServer` 甚至有几分相似（都有 `newConnection` 和 `removeConnection()` 这两个成员函数），只不过每个 `TcpClient` 只管理一个 `TcpConnection`。\n",
    "\n",
    "- [TcpClient.h](./12_tcpclient/tcp/TcpClient.h)\n",
    "\n",
    "代码从略，此处谈几个要点：\n",
    "\n",
    "- `TcpClient` 具备 `TcpConnection` 断开之后重新连接的功能，加上 `Connector` 具备反复尝试连接的功能，因此客户端和服务器端的启动顺序无关紧要。可以先启动客户端，一旦服务器启动，半分钟之内即可恢复连接（由 `Connector::kMaxRetryDelayMs` 常数控制）；在客户端运行期间服务器可以重启，客户端也会自动重连。\n",
    "- 连接断开后初次重试的延迟应该有随机性，比如说服务器崩溃，它所有的客户连接同时断开，然后 `0.5s` 之后同时再次发起连接，这样既可能造成 `SYN` 丢包，也可能给服务器端带来短期大负载，影响其服务质量。因此每个 `TcpClient` 应该等待一段随机的时间（`0.5～2s`），再重试，避免拥塞。\n",
    "- 发起连接的时候如果发生 `TCP SYN` 丢包，那么系统默认的重试间隔是 `3s`，这期间不会返回错误码，而且这个间隔似乎不容易修改。如果需要缩短间隔，可以再用一个定时器，在 `0.5s` 或 `1s` 之后[发起另一次连接](https://bitsup.blogspot.com/2010/12/accelerated-connection-retry-for-http.html)。如果有需求的话，这个功能可以做到 `Connector` 中。\n",
    "- 目前本节实现的 `TcpClient` 没有充分测试动态增减的情况，也就是说没有充分测试 `TcpClient` 的生命周期比 `EventLoop` 短的情况，特别是没有充分测试 `TcpClient` 在连接建立期间析构的情况。编写这方面的单元测试多半要用到 §`12.4` 介绍的技术。\n",
    "\n",
    "注意目前 `muduo 0.8.0` 采用 `shared_ptr` 来管理 `Connector`，因为在编写这部分代码的时候 `TimerQueue` 尚不支持 `cancel()` 操作。将来 `muduo 1.0` 会在充分测试的前提下改用这里展示的简洁的实现。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# * 8.13 `epoll`\n",
    "\n",
    "`epoll(4)` 是 Linux 独有的高效的 IO multiplexing 机制，它与 `poll(2)` 的不同之处主要在于 `poll(2)` 每次返回整个文件描述符数组，用户代码需要遍历数组以找到哪些文件描述符上有 IO 事件（见 p.285 的 `Poller::fillActiveChannels()`），而 `epoll_wait(2)` 返回的是活动 `fd` 的列表，需要遍历的数组通常会小得多。在并发连接数较大而活动连接比例不高时，`epoll(4)` 比 `poll(2)` 更高效。\n",
    "\n",
    "本节我们把 `epoll(4)` 封装为 `EPoller` class，它与 § 8.1.2 的 `Poller` class 具有完全相同的接口。`muduo` 实际的做法是定义 `Poller` 基类并提供两份实现 `PollPoller` 和 `EPollPoller`。这里为了简单起见，我们直接修改 `EventLoop`，只需把代码中的 `Poller` 替换为 `EPoller`。\n",
    "\n",
    "`EPoller` 的关键数据结构如下，其中 `events_` 不是保存所有关注的 `fd` 列表，而是一次 `epoll_wait(2)` 调用返回的活动 `fd` 列表，它的大小是自适应的。\n",
    "\n",
    "- [`events_`](./13_epoll/reactor/EPoller.h#L61)\n",
    "\n",
    "`struct epoll_event` 的定义如下注意，`epoll_data` 是个 `union`，muduo 使用的是其 `ptr` 成员，用于存放 [`Channel*`](./13_epoll/reactor/EPoller.h#L57)，这样可以减少一步 look up。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// /usr/include/x86_64-linux-gnu/sys/epoll.h\n",
    "\n",
    "typedef union epoll_data\n",
    "{\n",
    "  void *ptr;\n",
    "  int fd;\n",
    "  uint32_t u32;\n",
    "  uint64_t u64;\n",
    "} epoll_data_t;\n",
    "\n",
    "struct epoll_event\n",
    "{\n",
    "  uint32_t events;\t/* Epoll events */\n",
    "  epoll_data_t data;\t/* User data variable */\n",
    "} __EPOLL_PACKED;"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "为了减少转换，muduo `Channel` 没有自己定义 IO 事件的常量，而是直接使用 `poll(2)` 的定义（`POLLIN`、`POLLOUT` 等等），在 Linux 中它们和 `epoll(4)` 的常量相等。\n",
    "\n",
    "`EPoller::poll()` 的关键代码如下。`L58: &*events_.begin()` 在 `C++11` 中可写为 `events_.data()`。`L68` 表示如果当前活动 `fd` 的数目填满了 `events_`，那么下次就尝试收集更多的活动 `fd`。`events_` 的初始长度是 `16 (kInitEventListSize)`，其会根据程序的 IO 繁忙程度自动增长，但目前不会自动收缩。\n",
    "\n",
    "- [`EPoller::poll()`](./13_epoll/reactor/EPoller.cc#L55-L81)\n",
    "\n",
    "此处`epoll_wait(2)`的错误处理从略。\n",
    "\n",
    "- [`EPoller::poll()`](./13_epoll/reactor/EPoller.cc#L77)\n",
    "\n",
    "`EPoller::fillActiveChannels()` 的功能是将 `events_` 中的活动 `fd` 填入 `activeChannels`，其中 [`L90~L93`](./13_epoll/reactor/EPoller.cc#L90-L94) 是在检查 `invariant`。\n",
    "\n",
    "- [`EPoller::fillActiveChannels()`](./13_epoll/reactor/EPoller.cc#L82-L99)\n",
    "\n",
    "`updateChannel()` 和 `removeChannel()` 的代码从略。因为 `epoll` 是有状态的，因此这两个函数要时刻维护内核中的 `fd` 状态与应用程序的状态相符，`Channel::index()` 和 `Channel::set_index()` 被挪用为标记此 `Channel` 是否位于 `epoll` 的关注列表之中。这两个函数的复杂度是 `O(log N)`，因为 Linux 内核用红黑树来管理 `epoll` 关注的文件描述符清单。\n",
    "\n",
    "测试程序无需修改，全都已经自动用上了 `epoll(4)`。\n",
    "\n",
    "至此，一个基于事件的非阻塞 TCP 网络库已经初具规模。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 8.14 测试程序一览\n",
    "\n",
    "本章简要介绍了 `muduo` 的实现过程，是一个具有教学示范意义的项目，希望有助于读者理解 `one loop per thread` 这一编程模型背后的实现，在运用时更加得心应手。如果对本章代码有疑问，应该以最新版的 `muduo` 源代码为准。\n",
    "\n",
    "本节没有配套代码，以下列出前面各节出现的测试代码的功能。\n",
    "\n",
    "- § 8.0 s00/test1.cc 在两个线程里各自运行一个 `EventLoop`。\n",
    "- § 8.0 s00/test2.cc 试图在非 IO 线程调用 `EventLoop::loop()`，程序崩溃。\n",
    "- § 8.1 s01/test3.cc 用 `Channel` 关注 `timerfd` 的可读事件。\n",
    "- § 8.2 s02/test4.cc `TimerQueue` 示例。\n",
    "- § 8.3 s03/test5.cc IO 线程调用 `EventLoop::runInLoop()` 和 `EventLoop::runAfter()`。\n",
    "- § 8.3 s03/test6.cc 跨线程调用 `EventLoop::runInLoop()` 和 `EventLoop::runAfter()`。\n",
    "- § 8.4 s04/test7.cc `Acceptor` 示例。\n",
    "- § 8.5 s05/test8.cc `discard` 服务。\n",
    "- § 8.8 s08/test9.cc `echo` 服务。\n",
    "- § 8.8 s08/test10.cc 发送两次数据，测试 `TcpConnection::send()`。\n",
    "- § 8.9 s09/test11.cc `chargen` 服务，使用 `WriteCompleteCallback`。\n",
    "- § 8.11 s11/test12.cc `Connector` 示例。\n",
    "- § 8.12 s12/test13.cc `TcpClient` 示例。\n",
    "\n",
    "本章 `Acceptor`、`Connector`、`Reactor` 等术语是 Douglas Schmidt 发明的，他的原始论文出处是\n",
    "\n",
    "- http://www.cs.wustl.edu/~schmidt/PDF/Reactor1-93.pdf\n",
    "- http://www.cs.wustl.edu/~schmidt/PDF/Reactor2-93.pdf\n",
    "- http://www.cs.wustl.edu/~schmidt/PDF/reactor-siemens.pdf\n",
    "- http://www.cs.wustl.edu/~schmidt/PDF/reactor-rules.pdf\n",
    "- http://www.cs.wustl.edu/~schmidt/PDF/Acceptor.pdf\n",
    "- http://www.cs.wustl.edu/~schmidt/PDF/Connector.pdf\n",
    "- http://www.cs.wustl.edu/~schmidt/PDF/Acc-Con.pdf\n",
    "\n",
    "我在一篇[访谈](http://www.oschina.net/question/2861182) 中谈到了 `muduo` 将来的计划：`1.0` 版完善单元测试，基本覆盖各种 `code path`，特别是各种 `Sockets API` 出错情况的测试，以及用户调用与 IO 事件的交互。`2.0` 版启用 `C++11`，特别是 `rvalue reference` 有助于提高性能与资源管理的便利性。以上计划中的版本尚无明确的时间表。"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "C++17",
   "language": "C++17",
   "name": "xcpp17"
  },
  "language_info": {
   "codemirror_mode": "text/x-c++src",
   "file_extension": ".cpp",
   "mimetype": "text/x-c++src",
   "name": "C++17",
   "version": "17"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
